sync_unix.h

fastdb-2.92的源码

                        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

#define USE_LOCAL_CS_IMPL

// 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__