sync_unix.h

一个不错的fastdb使用例子

        if (errno == ENOENT)
        {
          sem = sem_open(name, O_CREAT|O_EXCL, 0777, 0);

          if (sem != SEM_FAILED)
          {
            status = NotYetInitialized;
            break;
          }
          else if (errno != EEXIST)
          {
            status = InitializationError;
            break;
          }
        }
        else
        {
          status = InitializationError;
          break;
        }
      }
      else
      {
        status = (sem_wait(sem) == 0 && sem_post(sem) == 0)
                 ? AlreadyInitialized : InitializationError;
        break;
      }
    }

    delete[] tmp;
    return status;
  }

  void done()
  {
    sem_post(sem);
  }

  bool close()
  {
    sem_close(sem);
    return false;
  }

  void erase()
  {
    close();
  }
};

class dbSemaphore
{

protected:
  sem_t* sem;

public:
  void wait()
  {
    int rc = sem_wait(sem);
    assert(rc == 0);
  }

  bool wait(unsigned msec)
  {
#ifdef POSIX_1003_1d

    struct timespec abs_ts;

    struct timeval  cur_tv;
    clock_gettime(CLOCK_REALTIME, &cur_tv);
    abs_ts.tv_sec = cur_tv.tv_sec + (msec + cur_tv.tv_usec / 1000) / 1000000;
    abs_ts.tv_nsec = (msec + cur_tv.tv_usec / 1000) % 1000000 * 1000;
    int rc = sem_timedwait(sem, &abs_ts);

    if (rc < 0)
    {
      assert(errno == ETIMEDOUT);
      return false;
    }

    return true;
#else

    int rc = sem_wait(sem);
    assert(rc == 0);
    return true;
#endif

  }

  void signal(unsigned inc = 1)
  {
    while (--inc > 0)
    {
      sem_post(sem);
    }
  }

  void reset()
  {
    while (sem_trywait(sem) == 0)

      ;
  }

  bool open(char const* name, unsigned initValue = 0)
  {
    char* tmp = NULL;

    if (*name != '/')
    {
      tmp = new char[strlen(name)+2];
      strcpy(tmp+1, name);
      *tmp = '/';
      name = tmp;
    }

    sem = sem_open(name, O_CREAT, 0777, initValue);
    delete[] tmp;
    return sem != NULL;
  }

  void close()
  {
    sem_close(sem);
  }

  void erase()
  {
    close();
  }
};

class dbEvent : public dbSemaphore
{

public:
  void wait()
  {
    dbSemaphore::wait();
    sem_post(sem);
  }

  bool wait(unsigned msec)
  {
    if (dbSemaphore::wait(msec))
    {
      sem_post(sem);
      return true;
    }

    return false;
  }

  void signal()
  {
    while (sem_trywait(sem) == 0)

      ;
    sem_post(sem);
  }

  void reset()
  {
    while (sem_trywait(sem) == 0)

      ;
  }

  bool open(char const* name, bool signaled = false)
  {
    return dbSemaphore::open(name, (int)signaled);
  }
};

#else // USE_POSIX_SEMAPHORES

// Define local implemenation of InitializationMutex in sync.cpp

class dbInitializationMutex
{
  int semid;

public:
  enum initializationStatus {
    InitializationError,
    AlreadyInitialized,
    NotYetInitialized
  };
  initializationStatus initialize(char const* name);
  void done();
  bool close();
  void erase();
};


class dbSemaphore
{
  int s;

public:
  bool wait(unsigned msec = INFINITE);
  void signal(unsigned inc = 1);
  bool open(char const* name, unsigned initValue = 0);
  void reset();
  void close();
  void erase();
};

class dbEvent
{
  int e;

public:
  bool wait(unsigned msec = INFINITE);
  void signal();
  void reset();
  bool open(char const* name, bool signaled = false);
  void close();
  void erase();
};

#endif // USE_POSIX_SEMAPHORES


// Define dbSharedObject and dbSharedMemory
#if defined(USE_POSIX_MMAP) && USE_POSIX_MMAP

// For POSIX dbSharedObject, we use mmap()
template<class T>

class dbSharedObject
{
  char* name;
  T*  ptr;
  int fd;

public:

  dbSharedObject()
  {
    name = NULL;
    ptr = NULL;
    fd = -1;
  }

  bool open(char* fileName)
  {
    delete[] name;
    name = new char[strlen(fileName) + 1];
    strcpy(name, fileName);
    fd = ::open(fileName, O_RDWR|O_CREAT, 0777);

    if (fd < 0)
    {
      return false;
    }

    if (ftruncate(fd, sizeof(T)) < 0)
    {
      ::close(fd);
      return false;
    }

    ptr = (T*)mmap(NULL,
                   DOALIGN(sizeof(T), 4096),
                   PROT_READ|PROT_WRITE,
                   MAP_SHARED,
                   fd,
                   0);

    if (ptr == MAP_FAILED)
    {
      ptr = NULL;
      ::close(fd);
      return false;
    }

    return true;
  }

  T* get
    ()
  {
    return ptr;
  }

  void close()
  {
    if (ptr != NULL)
    {
      munmap(ptr, DOALIGN(sizeof(T), 4096));
    }

    if (fd > 0)
    {
      ::close(fd);
    }
  }

  void erase()
  {
    close();
    unlink(name);
  }

  ~dbSharedObject()
  {
    delete[] name;
  }
};

#else // USE_POSIX_MMAP

// Non POSIX, internal implementations of SharedMemory and SharedObject
extern char const* keyFileDir; // default value: "/tmp/"

class dbSharedMemory
{

protected:
  char*  ptr;
  int    shm;

public:
  bool  open(char const* name, size_t size);
  void  close();
  void  erase();
  char* get_base()
  {
    return ptr;
  }
};

template<class T>

class dbSharedObject : public dbSharedMemory
{

public:
  bool open(char* name)
  {
    return dbSharedMemory::open(name, sizeof(T));
  }

  T* get
    ()
  {
    return (T*)ptr;
  }
};

#endif

//////////////////////////////////////////////////////////////////////////
// Define dBGlobalCriticalSection for various platforms

// QNX uses a pthread based mutex for its implementation
//     Use only if pthread support is also enabled, else we'll use the default case
#if defined(__QNX__) && !defined(NO_PTHREADS)
typedef pthread_mutex_t sharedsem_t;

class dbGlobalCriticalSection
{
  pthread_mutexattr_t attr;
  sharedsem_t* sem;

public:
  void enter()
  {
    int rc = pthread_mutex_lock(sem);
    assert(rc == 0);
  }

  void leave()
  {
    int rc = pthread_mutex_unlock(sem);
    assert(rc == 0);
  }

  bool open(char const*, sharedsem_t* shr)
  {
    sem = shr;
    return true;
  }

  bool create(char const*, sharedsem_t* shr)
  {
    sem = shr;
    pthread_mutexattr_init(&attr);
    pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
    pthread_mutexattr_setrecursive(&attr, PTHREAD_RECURSIVE_ENABLE);
    pthread_mutex_init(sem, &attr);
    return true;
  }

  void close()
  {}

  void erase()
  {
    pthread_mutex_destroy(sem);
  }
};


#elif defined(__osf__)
// OSF uses "shared memory semaphores", located within a region mapped with mmap().
// XXX: Perhaps Konstantin can give some insight why this method was chosen
#include <errno.h>
typedef msemaphore sharedsem_t;

class dbGlobalCriticalSection
{
  sharedsem_t* sem;

public:
  void enter()
  {
    int rc;

    while ((rc = msem_lock(sem, 0)) < 0 && errno == EINTR)

      ;
    assert(rc == 0);
  }

  void leave()
  {
    int rc = msem_unlock(sem, 0);
    assert(rc == 0);
  }

  bool open(char const*, sharedsem_t* shr)
  {
    sem = shr;
    return true;
  }

  bool create(char const*, sharedsem_t* shr)
  {
    sem = shr;
    msem_init(shr, MSEM_UNLOCKED);
    return true;
  }

  void close()
  {}

  void erase()
  {
    msem_remove(sem);
  }
};


#elif defined(__sun)
// Sun uses the Solaris style semaphore implemenation (sema_init(), sema_post())
#include <synch.h>
#include <errno.h>
typedef sema_t sharedsem_t;

class dbGlobalCriticalSection
{
  sharedsem_t* sem;

public:
  void enter()
  {
    int rc;

    while ((rc = sema_wait(sem)) < 0 && errno == EINTR)

      ;
    assert(rc == 0);
  }

  void leave()
  {
    int rc = sema_post(sem);
    assert(rc == 0);
  }

  bool open(char const*, sharedsem_t* shr)
  {
    sem = shr;
    return true;
  }

  bool create(char const*, sharedsem_t* shr)
  {
    sem = shr;
    return sema_init(shr, 1, USYNC_PROCESS, NULL) == 0;
  }

  void close()
  {}

  void erase()
  {
    sema_destroy(sem);
  }
};

#elif defined(USE_POSIX_SEMAPHORES)
// Everyone else uses the POSIX style semaphores (sem_wait(), sem_post(), etc) if defined
typedef sem_t sharedsem_t;

class dbGlobalCriticalSection
{
  sharedsem_t* sem;

public:
  void enter()
  {
    int rc = sem_wait(sem);
    assert(rc == 0);
  }

  void leave()
  {
    int rc = sem_post(sem);
    assert(rc == 0);
  }

  bool open(char const* name, sharedsem_t* shr)
  {
    sem = shr;
    return true;
  }

  // XXX: BUG Check the sem_init() return
  bool create(char const* name, sharedsem_t* shr)
  {
    sem = shr;
    return sem_init(sem, 1, 1) == 0;
  }

  void close()
  {}

  void erase()
  {
    sem_destroy(sem);
  }
};

#else
// Lastly, use the local implementation
typedef long sharedsem_t;

class dbGlobalCriticalSection
{
  int          semid;
  sharedsem_t* count;

public:
  void enter();
  void leave();
  bool open(char const* name, sharedsem_t* shr);
  bool create(char const* name, sharedsem_t* shr);
  void close()
  {}

  void erase();
};

#endif //dbGLobalCriticalSection switch

#endif //__SYNC_UNIX_H__