[阶段1-B2] libtask 自研实现 — 定时器/事件/消息队列三大子系统
基于 SPEC §8 完整实现 (21 API): - Timer: create/start/stop/restart/destroy/is_active, 独立轮询线程10ms - Event: create/destroy/send/recv(AND|OR)/clear/query, 位图+cond同步 - MsgQueue: create/destroy/send/send_timeout/recv/try_recv/count/space, 环形缓冲 - C接口 + extern "C", C++内部 pthread, test: Timer=3 Event=0x5 Msg=hi
This commit is contained in:
parent
31c0e70063
commit
f2a099e4b9
|
|
@ -0,0 +1,143 @@
|
|||
/**
|
||||
* @file myTask.h
|
||||
* @brief Task framework: timers + events + message queues (C interface)
|
||||
* @details Based on SPEC §8 analysis. Uses C++ internally.
|
||||
*/
|
||||
|
||||
#ifndef _MY_TASK_H_
|
||||
#define _MY_TASK_H_
|
||||
#include <stdint.h>
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C"
|
||||
{
|
||||
#endif
|
||||
|
||||
/* ================================================================
|
||||
* Opaque handles
|
||||
* ================================================================ */
|
||||
|
||||
typedef struct stru_task_timer *stru_task_timer_t;
|
||||
typedef struct stru_task_event *stru_task_event_t;
|
||||
typedef struct stru_task_msg_queue *stru_task_msg_queue_t;
|
||||
|
||||
/* ================================================================
|
||||
* Timer callback — returns 0 to continue, non-zero to stop
|
||||
* ================================================================ */
|
||||
|
||||
typedef int (*timer_func_cb)(void *arg);
|
||||
|
||||
/* Timer flags */
|
||||
#define TIMER_ONCE 0 /* one-shot */
|
||||
#define TIMER_PERIOD 1 /* periodic */
|
||||
|
||||
/* ================================================================
|
||||
* Timer API (SPEC T01-T07)
|
||||
* ================================================================ */
|
||||
|
||||
/**
|
||||
* @brief Create a timer (does not start automatically)
|
||||
* @param name timer name for debugging
|
||||
* @param cb callback function
|
||||
* @param arg user argument passed to callback
|
||||
* @param timeout_ms interval in milliseconds
|
||||
* @param flags TIMER_ONCE or TIMER_PERIOD
|
||||
* @return timer handle, NULL on failure
|
||||
*/
|
||||
stru_task_timer_t task_timer_create(const char *name, timer_func_cb cb,
|
||||
void *arg, uint32_t timeout_ms, int flags);
|
||||
|
||||
/** Start the timer */
|
||||
int task_timer_start(stru_task_timer_t pt);
|
||||
|
||||
/** Stop the timer */
|
||||
int task_timer_stop(stru_task_timer_t pt);
|
||||
|
||||
/** Restart with new interval */
|
||||
int task_timer_restart(stru_task_timer_t pt, uint32_t timeout_ms);
|
||||
|
||||
/** Destroy and free the timer */
|
||||
int task_timer_destroy(stru_task_timer_t pt);
|
||||
|
||||
/** Check if timer is currently active */
|
||||
int task_timer_is_active(stru_task_timer_t pt);
|
||||
|
||||
/* ================================================================
|
||||
* Event API (SPEC T08-T13)
|
||||
* ================================================================ */
|
||||
|
||||
#define EVENT_OPT_AND 0 /* wait for ALL bits set */
|
||||
#define EVENT_OPT_OR 1 /* wait for ANY bit set */
|
||||
|
||||
/**
|
||||
* @brief Create an event object
|
||||
* @param name event name for debugging
|
||||
* @return event handle, NULL on failure
|
||||
*/
|
||||
stru_task_event_t task_event_create(const char *name);
|
||||
|
||||
/** Destroy event object */
|
||||
int task_event_destroy(stru_task_event_t pe);
|
||||
|
||||
/** Set (send) event bits */
|
||||
int task_event_send(stru_task_event_t pe, uint32_t bits);
|
||||
|
||||
/**
|
||||
* @brief Wait for event bits
|
||||
* @param pe event handle
|
||||
* @param set bits to wait for
|
||||
* @param opt EVENT_OPT_AND or EVENT_OPT_OR
|
||||
* @param timeout_ms timeout in ms (0 = forever)
|
||||
* @param recved [out] bits actually received
|
||||
* @return 0 on success, -1 on timeout
|
||||
*/
|
||||
int task_event_recv(stru_task_event_t pe, uint32_t set, uint32_t opt,
|
||||
uint32_t timeout_ms, uint32_t *recved);
|
||||
|
||||
/** Clear specific event bits */
|
||||
int task_event_clear(stru_task_event_t pe, uint32_t bits);
|
||||
|
||||
/** Query current event bits (non-blocking) */
|
||||
uint32_t task_event_query(stru_task_event_t pe);
|
||||
|
||||
/* ================================================================
|
||||
* Message Queue API (SPEC T14-T21)
|
||||
* ================================================================ */
|
||||
|
||||
/**
|
||||
* @brief Create a fixed-size message queue
|
||||
* @param name queue name
|
||||
* @param msg_size max size of each message (bytes)
|
||||
* @param msg_num max number of messages
|
||||
* @return queue handle, NULL on failure
|
||||
*/
|
||||
stru_task_msg_queue_t task_msg_queue_create(const char *name,
|
||||
uint32_t msg_size, uint32_t msg_num);
|
||||
|
||||
/** Destroy message queue */
|
||||
int task_msg_queue_destroy(stru_task_msg_queue_t pq);
|
||||
|
||||
/** Send message (blocking if queue full) */
|
||||
int task_msg_queue_send(stru_task_msg_queue_t pq, const void *msg, uint32_t size);
|
||||
|
||||
/** Send message with timeout (ms), returns -1 on timeout */
|
||||
int task_msg_queue_send_timeout(stru_task_msg_queue_t pq, const void *msg,
|
||||
uint32_t size, uint32_t timeout_ms);
|
||||
|
||||
/** Receive message with timeout (ms), returns -1 on timeout */
|
||||
int task_msg_queue_recv(stru_task_msg_queue_t pq, void *msg, uint32_t size,
|
||||
uint32_t timeout_ms);
|
||||
|
||||
/** Try receive (non-blocking), returns 0 on success, -1 if empty */
|
||||
int task_msg_queue_try_recv(stru_task_msg_queue_t pq, void *msg, uint32_t size);
|
||||
|
||||
/** Get number of messages currently in queue */
|
||||
uint32_t task_msg_queue_get_count(stru_task_msg_queue_t pq);
|
||||
|
||||
/** Get free space in queue */
|
||||
uint32_t task_msg_queue_space(stru_task_msg_queue_t pq);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
#endif
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
include ./../../../linux.mk
|
||||
L := $(notdir $(realpath $(CURDIR)/..))
|
||||
M := libtask
|
||||
O := $(LIB_REL)/$(M).a
|
||||
S := $(SRC_ROOT_DIR)/$(L)/libtask/src
|
||||
I := -I$(SRC_ROOT_DIR)/$(L)/libtask/inc
|
||||
B := $(CURDIR)/$(M)/obj
|
||||
SRCS := $(wildcard $(S)/*.cpp)
|
||||
OBJS := $(patsubst $(S)/%.cpp,$(B)/%.o,$(SRCS))
|
||||
F := $(CXX_FLAGS) $(I) -I$(SRC_ROOT_DIR)/public/liblog/inc
|
||||
all:
|
||||
@mkdir -p $(ROOT_DIR)/release/inc
|
||||
@cp -f $(S:src=inc)/*.h $(ROOT_DIR)/release/inc/ 2>/dev/null; true
|
||||
@$(MAKE) $(O)
|
||||
$(O): $(OBJS)
|
||||
@mkdir -p $(dir $@)
|
||||
$(AR) rcs $@ $^
|
||||
@echo "[$(M)] built"
|
||||
$(B)/%.o: $(S)/%.cpp
|
||||
@mkdir -p $(dir $@)
|
||||
$(CXX) $(F) -c $< -o $@
|
||||
clean:
|
||||
rm -rf $(B) $(O)
|
||||
rebuild: clean all
|
||||
.PHONY: all clean rebuild
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
SUBDIRS := ./liblist ./libfunc ./liblog ./libmd5 ./libcJSON ./libmy_xxhash
|
||||
SUBDIRS := ./liblist ./libfunc ./liblog ./libmd5 ./libcJSON ./libmy_xxhash ./libtask
|
||||
define make_subdir
|
||||
@for d in $(SUBDIRS); do [ -f "$$d/makefile" ] && (cd "$$d" && make -f makefile $1) || true; done
|
||||
endef
|
||||
|
|
|
|||
|
|
@ -0,0 +1,143 @@
|
|||
/**
|
||||
* @file myTask.h
|
||||
* @brief Task framework: timers + events + message queues (C interface)
|
||||
* @details Based on SPEC §8 analysis. Uses C++ internally.
|
||||
*/
|
||||
|
||||
#ifndef _MY_TASK_H_
|
||||
#define _MY_TASK_H_
|
||||
#include <stdint.h>
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C"
|
||||
{
|
||||
#endif
|
||||
|
||||
/* ================================================================
|
||||
* Opaque handles
|
||||
* ================================================================ */
|
||||
|
||||
typedef struct stru_task_timer *stru_task_timer_t;
|
||||
typedef struct stru_task_event *stru_task_event_t;
|
||||
typedef struct stru_task_msg_queue *stru_task_msg_queue_t;
|
||||
|
||||
/* ================================================================
|
||||
* Timer callback — returns 0 to continue, non-zero to stop
|
||||
* ================================================================ */
|
||||
|
||||
typedef int (*timer_func_cb)(void *arg);
|
||||
|
||||
/* Timer flags */
|
||||
#define TIMER_ONCE 0 /* one-shot */
|
||||
#define TIMER_PERIOD 1 /* periodic */
|
||||
|
||||
/* ================================================================
|
||||
* Timer API (SPEC T01-T07)
|
||||
* ================================================================ */
|
||||
|
||||
/**
|
||||
* @brief Create a timer (does not start automatically)
|
||||
* @param name timer name for debugging
|
||||
* @param cb callback function
|
||||
* @param arg user argument passed to callback
|
||||
* @param timeout_ms interval in milliseconds
|
||||
* @param flags TIMER_ONCE or TIMER_PERIOD
|
||||
* @return timer handle, NULL on failure
|
||||
*/
|
||||
stru_task_timer_t task_timer_create(const char *name, timer_func_cb cb,
|
||||
void *arg, uint32_t timeout_ms, int flags);
|
||||
|
||||
/** Start the timer */
|
||||
int task_timer_start(stru_task_timer_t pt);
|
||||
|
||||
/** Stop the timer */
|
||||
int task_timer_stop(stru_task_timer_t pt);
|
||||
|
||||
/** Restart with new interval */
|
||||
int task_timer_restart(stru_task_timer_t pt, uint32_t timeout_ms);
|
||||
|
||||
/** Destroy and free the timer */
|
||||
int task_timer_destroy(stru_task_timer_t pt);
|
||||
|
||||
/** Check if timer is currently active */
|
||||
int task_timer_is_active(stru_task_timer_t pt);
|
||||
|
||||
/* ================================================================
|
||||
* Event API (SPEC T08-T13)
|
||||
* ================================================================ */
|
||||
|
||||
#define EVENT_OPT_AND 0 /* wait for ALL bits set */
|
||||
#define EVENT_OPT_OR 1 /* wait for ANY bit set */
|
||||
|
||||
/**
|
||||
* @brief Create an event object
|
||||
* @param name event name for debugging
|
||||
* @return event handle, NULL on failure
|
||||
*/
|
||||
stru_task_event_t task_event_create(const char *name);
|
||||
|
||||
/** Destroy event object */
|
||||
int task_event_destroy(stru_task_event_t pe);
|
||||
|
||||
/** Set (send) event bits */
|
||||
int task_event_send(stru_task_event_t pe, uint32_t bits);
|
||||
|
||||
/**
|
||||
* @brief Wait for event bits
|
||||
* @param pe event handle
|
||||
* @param set bits to wait for
|
||||
* @param opt EVENT_OPT_AND or EVENT_OPT_OR
|
||||
* @param timeout_ms timeout in ms (0 = forever)
|
||||
* @param recved [out] bits actually received
|
||||
* @return 0 on success, -1 on timeout
|
||||
*/
|
||||
int task_event_recv(stru_task_event_t pe, uint32_t set, uint32_t opt,
|
||||
uint32_t timeout_ms, uint32_t *recved);
|
||||
|
||||
/** Clear specific event bits */
|
||||
int task_event_clear(stru_task_event_t pe, uint32_t bits);
|
||||
|
||||
/** Query current event bits (non-blocking) */
|
||||
uint32_t task_event_query(stru_task_event_t pe);
|
||||
|
||||
/* ================================================================
|
||||
* Message Queue API (SPEC T14-T21)
|
||||
* ================================================================ */
|
||||
|
||||
/**
|
||||
* @brief Create a fixed-size message queue
|
||||
* @param name queue name
|
||||
* @param msg_size max size of each message (bytes)
|
||||
* @param msg_num max number of messages
|
||||
* @return queue handle, NULL on failure
|
||||
*/
|
||||
stru_task_msg_queue_t task_msg_queue_create(const char *name,
|
||||
uint32_t msg_size, uint32_t msg_num);
|
||||
|
||||
/** Destroy message queue */
|
||||
int task_msg_queue_destroy(stru_task_msg_queue_t pq);
|
||||
|
||||
/** Send message (blocking if queue full) */
|
||||
int task_msg_queue_send(stru_task_msg_queue_t pq, const void *msg, uint32_t size);
|
||||
|
||||
/** Send message with timeout (ms), returns -1 on timeout */
|
||||
int task_msg_queue_send_timeout(stru_task_msg_queue_t pq, const void *msg,
|
||||
uint32_t size, uint32_t timeout_ms);
|
||||
|
||||
/** Receive message with timeout (ms), returns -1 on timeout */
|
||||
int task_msg_queue_recv(stru_task_msg_queue_t pq, void *msg, uint32_t size,
|
||||
uint32_t timeout_ms);
|
||||
|
||||
/** Try receive (non-blocking), returns 0 on success, -1 if empty */
|
||||
int task_msg_queue_try_recv(stru_task_msg_queue_t pq, void *msg, uint32_t size);
|
||||
|
||||
/** Get number of messages currently in queue */
|
||||
uint32_t task_msg_queue_get_count(stru_task_msg_queue_t pq);
|
||||
|
||||
/** Get free space in queue */
|
||||
uint32_t task_msg_queue_space(stru_task_msg_queue_t pq);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
#endif
|
||||
|
|
@ -0,0 +1,302 @@
|
|||
/**
|
||||
* @file myTask.cpp
|
||||
* @brief Task framework implementation (C++ internal)
|
||||
*/
|
||||
|
||||
#include <unistd.h>
|
||||
#include "myTask.h"
|
||||
#include "myBase.h"
|
||||
#include <pthread.h>
|
||||
#include <map>
|
||||
#include <string>
|
||||
#include <sys/time.h>
|
||||
|
||||
struct stru_task_timer
|
||||
{
|
||||
timer_func_cb cb;
|
||||
void *arg;
|
||||
uint32_t ms;
|
||||
int flags;
|
||||
int active;
|
||||
uint64_t next_us;
|
||||
std::string name;
|
||||
};
|
||||
|
||||
LOCAL std::map<stru_task_timer_t, int> *g_timer_map = NULL;
|
||||
LOCAL pthread_t g_timer_thread;
|
||||
LOCAL int g_timer_running = 0;
|
||||
LOCAL pthread_mutex_t g_timer_lock = PTHREAD_MUTEX_INITIALIZER;
|
||||
|
||||
LOCAL uint64_t now_us(void)
|
||||
{
|
||||
struct timeval tv; gettimeofday(&tv, NULL);
|
||||
return (uint64_t)tv.tv_sec * 1000000 + (uint64_t)tv.tv_usec;
|
||||
}
|
||||
|
||||
LOCAL void *timer_worker(void *arg)
|
||||
{
|
||||
(void)arg;
|
||||
while (g_timer_running)
|
||||
{
|
||||
pthread_mutex_lock(&g_timer_lock);
|
||||
uint64_t now = now_us();
|
||||
for (auto &kv : *g_timer_map)
|
||||
{
|
||||
stru_task_timer_t pt = kv.first;
|
||||
if (!pt->active || now < pt->next_us) continue;
|
||||
int ret = pt->cb(pt->arg);
|
||||
if (ret != 0 || pt->flags == 0) pt->active = 0;
|
||||
else pt->next_us = now + (uint64_t)pt->ms * 1000;
|
||||
}
|
||||
pthread_mutex_unlock(&g_timer_lock);
|
||||
usleep(10000);
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
LOCAL void timer_ensure_init(void)
|
||||
{
|
||||
if (g_timer_map) return;
|
||||
g_timer_map = new std::map<stru_task_timer_t, int>();
|
||||
g_timer_running = 1;
|
||||
pthread_create(&g_timer_thread, NULL, timer_worker, NULL);
|
||||
}
|
||||
|
||||
struct stru_task_event
|
||||
{
|
||||
uint32_t bits; pthread_mutex_t lock; pthread_cond_t cond; std::string name;
|
||||
};
|
||||
|
||||
struct stru_task_msg_queue
|
||||
{
|
||||
uint8_t *buf; uint32_t msg_size, msg_num, head, tail, count;
|
||||
pthread_mutex_t lock; pthread_cond_t cond_read, cond_write; std::string name;
|
||||
};
|
||||
|
||||
/* ================================================================
|
||||
* Exported API (extern "C" — C linkage)
|
||||
* ================================================================ */
|
||||
|
||||
extern "C" {
|
||||
|
||||
stru_task_timer_t task_timer_create(const char *name, timer_func_cb cb,
|
||||
void *arg, uint32_t ms, int flags)
|
||||
{
|
||||
timer_ensure_init();
|
||||
stru_task_timer_t pt = new stru_task_timer();
|
||||
if (!pt) return NULL;
|
||||
pt->cb = cb; pt->arg = arg; pt->ms = ms; pt->flags = flags;
|
||||
pt->active = 0; pt->next_us = 0; pt->name = name ? name : "?";
|
||||
pthread_mutex_lock(&g_timer_lock);
|
||||
(*g_timer_map)[pt] = 1;
|
||||
pthread_mutex_unlock(&g_timer_lock);
|
||||
return pt;
|
||||
}
|
||||
|
||||
int task_timer_start(stru_task_timer_t pt)
|
||||
{
|
||||
if (!pt) return -1;
|
||||
pthread_mutex_lock(&g_timer_lock);
|
||||
pt->next_us = now_us() + (uint64_t)pt->ms * 1000;
|
||||
pt->active = 1;
|
||||
pthread_mutex_unlock(&g_timer_lock);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int task_timer_stop(stru_task_timer_t pt)
|
||||
{
|
||||
if (!pt) return -1;
|
||||
pthread_mutex_lock(&g_timer_lock); pt->active = 0; pthread_mutex_unlock(&g_timer_lock);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int task_timer_restart(stru_task_timer_t pt, uint32_t ms)
|
||||
{
|
||||
if (!pt) return -1;
|
||||
pthread_mutex_lock(&g_timer_lock);
|
||||
pt->ms = ms; pt->next_us = now_us() + (uint64_t)ms * 1000; pt->active = 1;
|
||||
pthread_mutex_unlock(&g_timer_lock);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int task_timer_destroy(stru_task_timer_t pt)
|
||||
{
|
||||
if (!pt) return -1;
|
||||
pthread_mutex_lock(&g_timer_lock); g_timer_map->erase(pt); pthread_mutex_unlock(&g_timer_lock);
|
||||
delete pt;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int task_timer_is_active(stru_task_timer_t pt)
|
||||
{
|
||||
if (!pt) return 0;
|
||||
pthread_mutex_lock(&g_timer_lock);
|
||||
int a = pt->active; pthread_mutex_unlock(&g_timer_lock);
|
||||
return a;
|
||||
}
|
||||
|
||||
stru_task_event_t task_event_create(const char *name)
|
||||
{
|
||||
stru_task_event_t pe = new stru_task_event();
|
||||
if (!pe) return NULL;
|
||||
pe->bits = 0;
|
||||
pe->lock = PTHREAD_MUTEX_INITIALIZER;
|
||||
pe->cond = PTHREAD_COND_INITIALIZER;
|
||||
pe->name = name ? name : "?";
|
||||
return pe;
|
||||
}
|
||||
|
||||
int task_event_destroy(stru_task_event_t pe)
|
||||
{
|
||||
if (!pe) return -1;
|
||||
pthread_mutex_destroy(&pe->lock); pthread_cond_destroy(&pe->cond); delete pe;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int task_event_send(stru_task_event_t pe, uint32_t bits)
|
||||
{
|
||||
if (!pe) return -1;
|
||||
pthread_mutex_lock(&pe->lock); pe->bits |= bits; pthread_cond_broadcast(&pe->cond); pthread_mutex_unlock(&pe->lock);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int task_event_recv(stru_task_event_t pe, uint32_t set, uint32_t opt,
|
||||
uint32_t timeout_ms, uint32_t *recved)
|
||||
{
|
||||
if (!pe || !recved) return -1;
|
||||
pthread_mutex_lock(&pe->lock);
|
||||
while (1)
|
||||
{
|
||||
uint32_t m = pe->bits & set;
|
||||
if ((opt == 0 && m == set) || (opt == 1 && m != 0))
|
||||
{ *recved = m; pe->bits &= ~set; pthread_mutex_unlock(&pe->lock); return 0; }
|
||||
if (timeout_ms == 0) { pthread_cond_wait(&pe->cond, &pe->lock); }
|
||||
else
|
||||
{
|
||||
struct timeval tv; gettimeofday(&tv, NULL);
|
||||
struct timespec ts;
|
||||
ts.tv_sec = tv.tv_sec + timeout_ms / 1000;
|
||||
ts.tv_nsec = (tv.tv_usec + (timeout_ms % 1000) * 1000) * 1000;
|
||||
if (ts.tv_nsec >= 1000000000L) { ts.tv_sec++; ts.tv_nsec -= 1000000000L; }
|
||||
if (pthread_cond_timedwait(&pe->cond, &pe->lock, &ts) == ETIMEDOUT)
|
||||
{ pthread_mutex_unlock(&pe->lock); return -1; }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
int task_event_clear(stru_task_event_t pe, uint32_t bits)
|
||||
{
|
||||
if (!pe) return -1;
|
||||
pthread_mutex_lock(&pe->lock); pe->bits &= ~bits; pthread_mutex_unlock(&pe->lock);
|
||||
return 0;
|
||||
}
|
||||
|
||||
uint32_t task_event_query(stru_task_event_t pe)
|
||||
{
|
||||
if (!pe) return 0;
|
||||
pthread_mutex_lock(&pe->lock);
|
||||
uint32_t b = pe->bits; pthread_mutex_unlock(&pe->lock);
|
||||
return b;
|
||||
}
|
||||
|
||||
stru_task_msg_queue_t task_msg_queue_create(const char *name,
|
||||
uint32_t msg_size, uint32_t msg_num)
|
||||
{
|
||||
if (msg_size == 0 || msg_num == 0) return NULL;
|
||||
stru_task_msg_queue_t pq = new stru_task_msg_queue();
|
||||
if (!pq) return NULL;
|
||||
pq->buf = (uint8_t *)malloc(msg_size * msg_num);
|
||||
pq->msg_size = msg_size; pq->msg_num = msg_num;
|
||||
pq->head = 0; pq->tail = 0; pq->count = 0;
|
||||
pq->lock = PTHREAD_MUTEX_INITIALIZER;
|
||||
pq->cond_read = PTHREAD_COND_INITIALIZER;
|
||||
pq->cond_write = PTHREAD_COND_INITIALIZER;
|
||||
pq->name = name ? name : "?";
|
||||
return pq;
|
||||
}
|
||||
|
||||
int task_msg_queue_destroy(stru_task_msg_queue_t pq)
|
||||
{
|
||||
if (!pq) return -1;
|
||||
pthread_mutex_destroy(&pq->lock); pthread_cond_destroy(&pq->cond_read); pthread_cond_destroy(&pq->cond_write);
|
||||
free(pq->buf); delete pq;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int task_msg_queue_send(stru_task_msg_queue_t pq, const void *msg, uint32_t size)
|
||||
{ return task_msg_queue_send_timeout(pq, msg, size, 0); }
|
||||
|
||||
int task_msg_queue_send_timeout(stru_task_msg_queue_t pq, const void *msg,
|
||||
uint32_t size, uint32_t timeout_ms)
|
||||
{
|
||||
if (!pq || !msg || size > pq->msg_size) return -1;
|
||||
pthread_mutex_lock(&pq->lock);
|
||||
while (pq->count >= pq->msg_num)
|
||||
{
|
||||
if (timeout_ms == 0) pthread_cond_wait(&pq->cond_write, &pq->lock);
|
||||
else {
|
||||
struct timeval tv; gettimeofday(&tv, NULL);
|
||||
struct timespec ts;
|
||||
ts.tv_sec = tv.tv_sec + timeout_ms / 1000;
|
||||
ts.tv_nsec = (tv.tv_usec + (timeout_ms % 1000) * 1000) * 1000;
|
||||
if (ts.tv_nsec >= 1000000000L) { ts.tv_sec++; ts.tv_nsec -= 1000000000L; }
|
||||
if (pthread_cond_timedwait(&pq->cond_write, &pq->lock, &ts) == ETIMEDOUT)
|
||||
{ pthread_mutex_unlock(&pq->lock); return -1; }
|
||||
}
|
||||
}
|
||||
memcpy(pq->buf + pq->head * pq->msg_size, msg, size);
|
||||
pq->head = (pq->head + 1) % pq->msg_num; pq->count++;
|
||||
pthread_cond_signal(&pq->cond_read); pthread_mutex_unlock(&pq->lock);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int task_msg_queue_recv(stru_task_msg_queue_t pq, void *msg, uint32_t size,
|
||||
uint32_t timeout_ms)
|
||||
{
|
||||
if (!pq || !msg) return -1;
|
||||
pthread_mutex_lock(&pq->lock);
|
||||
while (pq->count == 0)
|
||||
{
|
||||
if (timeout_ms == 0) pthread_cond_wait(&pq->cond_read, &pq->lock);
|
||||
else {
|
||||
struct timeval tv; gettimeofday(&tv, NULL);
|
||||
struct timespec ts;
|
||||
ts.tv_sec = tv.tv_sec + timeout_ms / 1000;
|
||||
ts.tv_nsec = (tv.tv_usec + (timeout_ms % 1000) * 1000) * 1000;
|
||||
if (ts.tv_nsec >= 1000000000L) { ts.tv_sec++; ts.tv_nsec -= 1000000000L; }
|
||||
if (pthread_cond_timedwait(&pq->cond_read, &pq->lock, &ts) == ETIMEDOUT)
|
||||
{ pthread_mutex_unlock(&pq->lock); return -1; }
|
||||
}
|
||||
}
|
||||
memcpy(msg, pq->buf + pq->tail * pq->msg_size, (size < pq->msg_size) ? size : pq->msg_size);
|
||||
pq->tail = (pq->tail + 1) % pq->msg_num; pq->count--;
|
||||
pthread_cond_signal(&pq->cond_write); pthread_mutex_unlock(&pq->lock);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int task_msg_queue_try_recv(stru_task_msg_queue_t pq, void *msg, uint32_t size)
|
||||
{
|
||||
if (!pq || !msg) return -1;
|
||||
pthread_mutex_lock(&pq->lock);
|
||||
if (pq->count == 0) { pthread_mutex_unlock(&pq->lock); return -1; }
|
||||
memcpy(msg, pq->buf + pq->tail * pq->msg_size, (size < pq->msg_size) ? size : pq->msg_size);
|
||||
pq->tail = (pq->tail + 1) % pq->msg_num; pq->count--;
|
||||
pthread_cond_signal(&pq->cond_write); pthread_mutex_unlock(&pq->lock);
|
||||
return 0;
|
||||
}
|
||||
|
||||
uint32_t task_msg_queue_get_count(stru_task_msg_queue_t pq)
|
||||
{
|
||||
if (!pq) return 0;
|
||||
pthread_mutex_lock(&pq->lock); uint32_t c = pq->count; pthread_mutex_unlock(&pq->lock);
|
||||
return c;
|
||||
}
|
||||
|
||||
uint32_t task_msg_queue_space(stru_task_msg_queue_t pq)
|
||||
{
|
||||
if (!pq) return 0;
|
||||
pthread_mutex_lock(&pq->lock); uint32_t s = pq->msg_num - pq->count; pthread_mutex_unlock(&pq->lock);
|
||||
return s;
|
||||
}
|
||||
|
||||
} /* extern "C" */
|
||||
Loading…
Reference in New Issue