update

wangzhengquan

2020-07-14 e4ce347428a0d459c7561f7bb3e180f351c11a4f

update

 queue/include/array_lock_free_queue.h

@@ -4,6 +4,7 @@
#include <assert.h> // assert()
#include <sched.h>  // sched_yield()
#include "logger_factory.h"
#include "mem_pool.h"

/// @brief implementation of an array based lock free queue with support for 
///        multiple producers
@@ -90,7 +91,7 @@
    ,m_count(0)           //
#endif
{
    m_theQueue = (ELEM_T*)mm_malloc(Q_SIZE * sizeof(ELEM_T));
    m_theQueue = (ELEM_T*)mem_pool_malloc(Q_SIZE * sizeof(ELEM_T));

}

@@ -98,7 +99,7 @@
ArrayLockFreeQueue<ELEM_T>::~ArrayLockFreeQueue()
{
    // std::cout << "destroy ArrayLockFreeQueue\n";
    mm_free(m_theQueue);
    mem_pool_free(m_theQueue);
    
}


 queue/include/mem_pool.h

@@ -1,15 +1,54 @@
#ifndef _MEM_POOL_H_
#define _MEM_POOL_H_  
#include "mm.h"
#include "sem_util.h"
#define MEM_POOL_COND_KEY 0x8801

static int mem_pool_cond  = SemUtil::get(MEM_POOL_COND_KEY, 0);

extern void mem_pool_init(size_t heap_size);
extern void mem_pool_destroy(void);
// static int mem_pool_mutex  = SemUtil::get(MEM_POOL_COND_KEY, 1);

extern void *mem_pool_malloc (size_t size);
extern void mem_pool_free (void *ptr);
extern void *mem_pool_realloc(void *ptr, size_t size);
extern hashtable_t * mem_pool_get_hashtable();
static inline void mem_pool_init(size_t heap_size) {
    if(mm_init(heap_size)) {
		 
    }
}

static inline void mem_pool_destroy(void) {
    if(mm_destroy()) {
        SemUtil::remove(mem_pool_cond);
    }
	
}

static inline void *mem_pool_malloc (size_t size) {
    void *ptr;
    while( (ptr = mm_malloc(size)) == NULL ) {
        err_msg(0, "There is not enough memery to allocate, waiting someone else to free.");
        SemUtil::set(mem_pool_cond, 0);
        // wait for someone else to free space
        SemUtil::dec(mem_pool_cond);

    }
	
    return ptr;
}

static inline void mem_pool_free (void *ptr) {
    mm_free(ptr);
    // notify malloc
    SemUtil::set(mem_pool_cond, 1);

}

static inline void *mem_pool_realloc (void *ptr, size_t size) {
    return mm_realloc(ptr, size);
}

static inline hashtable_t * mem_pool_get_hashtable() {
    return mm_get_hashtable();

}
// extern int mm_checkheap(int verbose);



 queue/include/mm.h

@@ -6,7 +6,7 @@
#include "hashtable.h"

extern bool mm_init(size_t heap_size);
extern void mm_destroy(void);
extern bool mm_destroy(void);

extern void *mm_malloc (size_t size);
extern void mm_free (void *ptr);

 queue/include/shm_queue_wrapper.h

@@ -9,13 +9,24 @@
#ifdef __cplusplus
extern "C" {
#endif
/**
 * 初始化共享内存
 * @size 共享内存大小
 * 
 */
void shm_init(int size);

/**
 * 销毁共享内存
 * 整个进程退出时需要执行这个方法，该方法首先会检查是否还有其他进程在使用该共享内存，如果还有其他进程在使用就只是detach,如果没有其他进程在使用则销毁整块内存。
 */
void shm_destroy();

/**
 * 创建队列
 * @ shmqueue 
 * @ key 标识共享队列的唯一key, 如果key为空会自动分配一个key，如果key不会空会检查是否有重复绑定的情况, 然后绑定key.
 * @ key 标识共享队列的唯一标识, key是一个指针里面存储了key的值， 如果key的值为-1系统会自动分配一个key值并把该key的值赋给key指针。如果key的值不会空会检查是否有重复绑定的情况, 有重复就报错没有就创建队列并绑定key.
 * @ queue_size 队列大小
 * @ ele_size 队列中元素大小
 */
void* shmqueue_create( int * key, int queue_size);


 queue/libshm_queue.a

Binary files differ

 queue/mem_pool.c

File was deleted

 queue/mm.c

@@ -374,7 +374,7 @@

 

void mm_destroy(void) {
bool mm_destroy(void) {
  struct shmid_ds shmid_ds;
  //detache
  SemUtil::dec(mutex);
@@ -395,13 +395,16 @@

       SemUtil::inc(mutex);
       SemUtil::remove(mutex);
       return true;

    } else {
      SemUtil::inc(mutex);
    }
    return false;
  } else {
    err_msg(errno, "mm_destroy shmctl IPC_STAT");
    SemUtil::inc(mutex);
    return false;
  }
 
  

 queue/shm_queue_wrapper.c

@@ -1,6 +1,6 @@
#include "shm_queue_wrapper.h"

#include "mm.h"
#include "mem_pool.h"
#include "hashtable.h"

typedef struct ele_t {
@@ -12,12 +12,20 @@
    void *mqueue;
} shmqueue_t;


void shm_init(int size) {
    mem_pool_init(size);
}

void shm_destroy() {
    mem_pool_destroy();
}

/**
 * 创建队列
 * @ shmqueue 
 * @ key 标识共享队列的唯一标识, key是一个指针里面存储了key的值， 如果key的值为-1系统会自动分配一个key值并把该key的值赋给key指针。如果key的值不会空会检查是否有重复绑定的情况, 有重复就报错没有就创建队列并绑定key.
 * @ queue_size 队列大小
 * @ size 队列中元素大小
 */
void* shmqueue_create( int * key, int queue_size) {
    int  mkey;
@@ -28,7 +36,7 @@
    } else {
        mkey = *key;
        if(hashtable_get(hashtable, mkey)!= NULL) {
            err_msg(0, "key %d has already been in used!", mkey);
            err_exit(0, "key %d has already been in used!", mkey);
            return NULL;
        }
    }
@@ -40,12 +48,12 @@
}

/**
 * 绑定key到队列，但是并不会创建队列。如果没有对应指定key的队列提示错误并返回空值
 * 绑定key到队列，但是并不会创建队列。如果没有对应指定key的队列提示错误并退出
 */
void* shmqueue_attach(int key) {
    hashtable_t *hashtable = mm_get_hashtable();
    if(hashtable_get(hashtable, key)== NULL) {
        err_msg(0, "shmqueue_attach：attach failed, The queue  binding on key %d has not been created!", key);
        err_exit(0, "shmqueue_attach：attach queue at key %d  failed!", key);
        return NULL;
    }

@@ -175,3 +183,5 @@
        return 0;
    }
}



 test/multiple_queue_consumer

Binary files differ

 test/multiple_queue_productor

Binary files differ

 test/single_consumer

Binary files differ

 test/single_productor

Binary files differ

 test/test_lostdata

Binary files differ

 test/test_queue

Binary files differ

 test/test_timeout

Binary files differ

 test2/Makefile

@@ -14,7 +14,7 @@
include $(ROOT)/Make.defines.$(PLATFORM)


PROGS =    test_queue_wrapper server client
PROGS =    test_queue_wrapper server client sem_test


build: $(PROGS)

 test2/client

Binary files differ

 test2/client.c

@@ -37,10 +37,10 @@

 
int main () {
    mm_init(512);
    shm_init(512);
    client();

    //整个进程退出时需要执行这个方法，该方法首先会检查是否还有其他进程在使用该共享内存，如果还有其他进程在使用就只是detach,如果没有其他进程在使用则销毁整块内存。
    mm_destroy();
    shm_destroy();
    return 0;
}

 test2/sem_test

Binary files differ

 test2/sem_test.c

New file
@@ -0,0 +1,21 @@
#include "usg_common.h"
#include "sem_util.h"
static int mem_pool_cond  = SemUtil::get(0x2001, 0);

void *run(void *varg) {
    sleep(5);
    // notify malloc
    SemUtil::set(mem_pool_cond, 1);
}

int main() {

  pthread_t tid;
  pthread_create(&tid, NULL, run, NULL); 
  SemUtil::set(mem_pool_cond, 0);
  // wait for someone else to free space
  printf("waiting...\n");
  SemUtil::dec(mem_pool_cond);
  printf("weak...\n");

}

 test2/server

Binary files differ

 test2/server.c

@@ -34,10 +34,10 @@

 
int main () {
    mm_init(512);
    shm_init(512);
    server();

    //整个进程退出时需要执行这个方法，该方法首先会检查是否还有其他进程在使用该共享内存，如果还有其他进程在使用就只是detach,如果没有其他进程在使用则销毁整块内存。
    mm_destroy();
    shm_destroy();
    return 0;
}

 test2/test_queue_wrapper

Binary files differ