diff --git a/release/inc/myTask.h b/release/inc/myTask.h new file mode 100644 index 0000000..0cfacaf --- /dev/null +++ b/release/inc/myTask.h @@ -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 + +#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 diff --git a/release/src/public/libtask/makefile b/release/src/public/libtask/makefile new file mode 100644 index 0000000..6d098e9 --- /dev/null +++ b/release/src/public/libtask/makefile @@ -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 diff --git a/release/src/public/makefile b/release/src/public/makefile index 8281a82..41daf40 100644 --- a/release/src/public/makefile +++ b/release/src/public/makefile @@ -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 diff --git a/src/public/libtask/inc/myTask.h b/src/public/libtask/inc/myTask.h new file mode 100644 index 0000000..0cfacaf --- /dev/null +++ b/src/public/libtask/inc/myTask.h @@ -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 + +#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 diff --git a/src/public/libtask/src/myTask.cpp b/src/public/libtask/src/myTask.cpp new file mode 100644 index 0000000..81e0be9 --- /dev/null +++ b/src/public/libtask/src/myTask.cpp @@ -0,0 +1,302 @@ +/** + * @file myTask.cpp + * @brief Task framework implementation (C++ internal) + */ + +#include +#include "myTask.h" +#include "myBase.h" +#include +#include +#include +#include + +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 *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(); + 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" */