refactor.

lichao

2021-06-02 ab898268c8bc493ca9862b2d64f2e1e7d20e5a4c

refactor.

 src/msg.h

@@ -22,8 +22,6 @@
#include "proto.h"
#include "shm.h"
#include <atomic>
#include <boost/interprocess/offset_ptr.hpp>
#include <functional>
#include <stdint.h>

class ShmSocket;
@@ -91,8 +89,6 @@

private:
    Meta *meta() const { return get<Meta>() - 1; }

    typedef std::function<void(void *p, int len)> ToArray;

    template <class Body>
    void *Pack(const BHMsgHead &head, const uint32_t head_len, const Body &body, const uint32_t body_len)

 src/shm.h

@@ -19,12 +19,10 @@
#ifndef SHM_6CHO6D6C
#define SHM_6CHO6D6C

#include "log.h"
#include <atomic>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <boost/noncopyable.hpp>
#include <thread>
#include <string>

namespace bhome_shm
{
@@ -32,67 +30,8 @@
using namespace boost::interprocess;

typedef managed_shared_memory mshm_t;

class MutexWithPidCheck
{
    typedef boost::interprocess::interprocess_mutex MutexT;
    static pid_t pid()
    {
        static pid_t val = getpid();
        return val;
    }
    static bool Killed(pid_t pid)
    {
        char buf[64] = {0};
        snprintf(buf, sizeof(buf) - 1, "/proc/%d/stat", pid);
        return access(buf, F_OK) != 0;
    }
    bool PidCas(pid_t exp, pid_t val) { return pid_.compare_exchange_strong(exp, val); }
    MutexT mutex_;
    std::atomic<pid_t> pid_;

public:
    typedef MutexT::internal_mutex_type internal_mutex_type;
    const internal_mutex_type &internal_mutex() const { return mutex_.internal_mutex(); }
    internal_mutex_type &internal_mutex() { return mutex_.internal_mutex(); }
    MutexWithPidCheck() :
        pid_(0) {}
    bool try_lock()
    {
        bool r = false;
        if (mutex_.try_lock()) {
            auto old = pid_.load();
            r = PidCas(old, pid());
        } else {
            auto old = pid_.load();
            if (Killed(old)) {
                r = PidCas(old, pid());
                if (r) {
                    LOG_DEBUG() << "PidCheck captured pid " << old << " -> " << pid();
                }
            }
        }
        return r;
    }

    void lock()
    {
        while (!try_lock()) {
            std::this_thread::yield();
        }
    }
    void unlock()
    {
        auto old = pid_.load();
        if (old == pid()) {
            mutex_.unlock();
        }
    }
};

typedef interprocess_mutex Mutex;
typedef scoped_lock<Mutex> Guard;
// typedef robust::Guard<Mutex> Guard;

class SharedMemory : public mshm_t
{
@@ -124,6 +63,9 @@
    {
        return construct<T>(name.c_str(), std::nothrow)(std::forward<decltype(params)>(params)...);
    }
    template <class T>
    bool Destroy(const std::string &name) { return destroy<T>(name.c_str()); }

    void *Alloc(const size_t size)
    {
        Guard lock(*pmutex_);
@@ -134,33 +76,10 @@
        Guard lock(*pmutex_);
        if (p) { deallocate(p); }
    }
    template <class T>
    void Dealloc(offset_ptr<T> ptr) { return Dealloc(ptr.get()); }

    template <class T, class... Params>
    T *New(Params &&...params)
    {
        Guard lock(*pmutex_);
        return construct<T>(anonymous_instance, std::nothrow)(std::forward<decltype(params)>(params)...);
    }
    template <class T>
    void Delete(T *p)
    {
        Guard lock(*pmutex_);
        if (p) { destroy_ptr<T>(p); };
    }
    template <class T>
    void Delete(offset_ptr<T> p) { Delete(p.get()); }
    template <class T>
    T *Find(const std::string &name) { return find<T>(name.c_str()).first; }
};

template <class D>
using Allocator = allocator<D, SharedMemory::segment_manager>;
template <class D>
using Deleter = deleter<D, SharedMemory::segment_manager>;
template <class D>
using SharedPtr = shared_ptr<D, Allocator<void>, Deleter<D>>;

enum Mode {
    eOpen = 0,
@@ -168,9 +87,9 @@
    eOpenOrCreate = 2
};

// ShmObject manages an object in shared memory, but ShmObject itself is not in shared memory.
// NamedShmObject manages an object in shared memory, but NamedShmObject itself is not in shared memory.
template <class T>
class ShmObject : private boost::noncopyable
class NamedShmObject : private boost::noncopyable
{
    static std::string ObjName(const std::string &name) { return "obj" + name; }

@@ -180,7 +99,7 @@
    ShmType &shm() const { return shm_; }

    template <class... Params>
    ShmObject(ShmType &segment, const std::string &name, Mode mode, Params &&...t) :
    NamedShmObject(ShmType &segment, const std::string &name, Mode mode, Params &&...t) :
        shm_(segment), name_(name)
    {
        switch (mode) {
@@ -200,7 +119,7 @@
        }
    }

    ShmObject(const int64_t offset, ShmType &segment, const std::string &name) :
    NamedShmObject(const int64_t offset, ShmType &segment, const std::string &name) :
        shm_(segment), name_(name)
    {
        pdata_ = reinterpret_cast<Data *>(Addr(shm_.get_address()) + offset);
@@ -208,10 +127,10 @@

    bool IsOk() const { return pdata_; }

    static bool Remove(SharedMemory &shm, const std::string &name) { return shm.destroy<Data>(ObjName(name).c_str()); }
    static bool Remove(SharedMemory &shm, const std::string &name) { return shm.Destroy<Data>(ObjName(name)); }
    static Data *Find(SharedMemory &shm, const std::string &name) { return shm.Find<Data>(ObjName(name)); }
    Data *Find(const std::string &name) { return Find(shm_, ObjName(name)); }
    virtual ~ShmObject() {}
    virtual ~NamedShmObject() {}
    std::string name() const { return name_; }
    Data *data() { return pdata_; }
    const Data *data() const { return pdata_; }

 src/shm_msg_queue.h

@@ -29,7 +29,7 @@
{
public:
    typedef int64_t RawData;
    typedef ShmObject<SharedQ63> Shmq;
    typedef NamedShmObject<SharedQ63> Shmq;
    typedef Shmq::Data Queue;
    typedef Shmq::ShmType ShmType;
    typedef uint64_t MQId;

 utest/robust_test.cpp

@@ -75,7 +75,7 @@
    int64_t d;
    BOOST_CHECK(tmp.pop(d));

    ShmObject<Rcb> rcb(shm, "test_rcb", eOpenOrCreate);
    NamedShmObject<Rcb> rcb(shm, "test_rcb", eOpenOrCreate);
    bool try_more = true;

    auto Writer = [&]() {

 utest/simple_tests.cpp

@@ -29,25 +29,11 @@
    SharedMemory shm(shm_name, 1024 * 1024 * 10);
    auto Avail = [&]() { return shm.get_free_memory(); };

    offset_ptr<const void> p;
    BOOST_CHECK(!p);
    BOOST_CHECK(p.get() == 0);
    p = 0;
    BOOST_CHECK(!p);
    BOOST_CHECK(p.get() == 0);
    const char *str = "basic";
    p = str;
    BOOST_CHECK(p);
    BOOST_CHECK(p.get() == str);
    p = 0;
    BOOST_CHECK(!p);
    BOOST_CHECK(p.get() == 0);

    auto init_avail = Avail();

    auto BasicTest = [&](int tid, int nloop) {
        auto Code = [&](int id) {
            typedef ShmObject<s1000> Int;
            typedef NamedShmObject<s1000> Int;
            std::string name = std::to_string(id);
            auto a0 = Avail();
            Int i1(shm, name, eOpenOrCreate);