diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index fa7c1f83bac34c54fefa210c2b3913bcc11db118..ef9b079cfc53c5975a6f217546ebdfcdca1247d7 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -295,6 +295,10 @@ void taos_close(TAOS *taos) { tscDebug("%p HB is freed", pHb); taosReleaseRef(tscObjRef, pHb->self); +#ifdef __APPLE__ + // to satisfy later tsem_destroy in taos_free_result + tsem_init(&pHb->rspSem, 0, 0); +#endif // __APPLE__ taos_free_result(pHb); } } diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 8dcb7425a8c4671dd46fe5e4679c896d1eaf00e0..e1521b4aa139f1d4cdf3c076c4a542d4a81cfb05 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -1933,6 +1933,10 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, __async_cb_func_t fp, void* param, in } if (tscAddSubqueryInfo(pCmd) != TSDB_CODE_SUCCESS) { +#ifdef __APPLE__ + // to satisfy later tsem_destroy in taos_free_result + tsem_init(&pNew->rspSem, 0, 0); +#endif // __APPLE__ tscFreeSqlObj(pNew); return NULL; } diff --git a/src/os/src/darwin/darwinSemphone.c b/src/os/src/darwin/darwinSemphone.c index d8e60cd2d55f4dc38834046bc8d7b702832ce8b1..a9e470bd7019c84d8c8c8368da242cbc4a1d618b 100644 --- a/src/os/src/darwin/darwinSemphone.c +++ b/src/os/src/darwin/darwinSemphone.c @@ -13,19 +13,76 @@ * along with this program. If not, see . */ +// fail-fast or let-it-crash philosophy +// https://en.wikipedia.org/wiki/Fail-fast +// https://stackoverflow.com/questions/4393197/erlangs-let-it-crash-philosophy-applicable-elsewhere +// experimentally, we follow log-and-crash here + #define _DEFAULT_SOURCE #include "os.h" +// #define SEM_USE_PTHREAD +// #define SEM_USE_POSIX +#define SEM_USE_SEM + +#ifdef SEM_USE_SEM +#include +#include +#include +#include + +static pthread_t sem_thread; +static pthread_once_t sem_once; +static task_t sem_port; +static volatile int sem_inited = 0; +static semaphore_t sem_exit; + +static void* sem_thread_routine(void *arg) { + (void)arg; + sem_port = mach_task_self(); + kern_return_t ret = semaphore_create(sem_port, &sem_exit, SYNC_POLICY_FIFO, 0); + if (ret != KERN_SUCCESS) { + fprintf(stderr, "==%s[%d]%s()==failed to create sem_exit\n", basename(__FILE__), __LINE__, __func__); + sem_inited = -1; + return NULL; + } + sem_inited = 1; + semaphore_wait(sem_exit); + return NULL; +} + +static void once_init(void) { + int r = 0; + r = pthread_create(&sem_thread, NULL, sem_thread_routine, NULL); + if (r) { + fprintf(stderr, "==%s[%d]%s()==failed to create thread\n", basename(__FILE__), __LINE__, __func__); + return; + } + while (sem_inited==0) { + ; + } +} +#endif + struct tsem_s { +#ifdef SEM_USE_PTHREAD + pthread_mutex_t lock; + pthread_cond_t cond; + volatile int64_t val; +#elif defined(SEM_USE_POSIX) + size_t id; + sem_t *sem; +#elif defined(SEM_USE_SEM) + semaphore_t sem; +#else // SEM_USE_PTHREAD dispatch_semaphore_t sem; +#endif // SEM_USE_PTHREAD + volatile unsigned int valid:1; }; -int tsem_wait(tsem_t *sem); -int tsem_post(tsem_t *sem); -int tsem_destroy(tsem_t *sem); int tsem_init(tsem_t *sem, int pshared, unsigned int value) { - fprintf(stderr, "==%s[%d]%s():[%p]==creating\n", basename(__FILE__), __LINE__, __func__, sem); + // fprintf(stderr, "==%s[%d]%s():[%p]==creating\n", basename(__FILE__), __LINE__, __func__, sem); if (*sem) { fprintf(stderr, "==%s[%d]%s():[%p]==already initialized\n", basename(__FILE__), __LINE__, __func__, sem); abort(); @@ -36,11 +93,61 @@ int tsem_init(tsem_t *sem, int pshared, unsigned int value) { abort(); } +#ifdef SEM_USE_PTHREAD + int r = pthread_mutex_init(&p->lock, NULL); + do { + if (r) break; + r = pthread_cond_init(&p->cond, NULL); + if (r) { + pthread_mutex_destroy(&p->lock); + break; + } + p->val = value; + } while (0); + if (r) { + fprintf(stderr, "==%s[%d]%s():[%p]==not created\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } +#elif defined(SEM_USE_POSIX) + static size_t tick = 0; + do { + size_t id = atomic_add_fetch_64(&tick, 1); + if (id==SEM_VALUE_MAX) { + atomic_store_64(&tick, 0); + id = 0; + } + char name[NAME_MAX-4]; + snprintf(name, sizeof(name), "/t%ld", id); + p->sem = sem_open(name, O_CREAT|O_EXCL, pshared, value); + p->id = id; + if (p->sem!=SEM_FAILED) break; + int e = errno; + if (e==EEXIST) continue; + if (e==EINTR) continue; + fprintf(stderr, "==%s[%d]%s():[%p]==not created[%d]%s\n", basename(__FILE__), __LINE__, __func__, sem, e, strerror(e)); + abort(); + } while (p->sem==SEM_FAILED); +#elif defined(SEM_USE_SEM) + pthread_once(&sem_once, once_init); + if (sem_inited!=1) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal resource init failed\n", basename(__FILE__), __LINE__, __func__, sem); + errno = ENOMEM; + return -1; + } + kern_return_t ret = semaphore_create(sem_port, &p->sem, SYNC_POLICY_FIFO, 0); + if (ret != KERN_SUCCESS) { + fprintf(stderr, "==%s[%d]%s():[%p]==semophore_create failed\n", basename(__FILE__), __LINE__, __func__, sem); + // we fail-fast here, because we have less-doc about semaphore_create for the moment + abort(); + } +#else // SEM_USE_PTHREAD p->sem = dispatch_semaphore_create(value); if (p->sem == NULL) { fprintf(stderr, "==%s[%d]%s():[%p]==not created\n", basename(__FILE__), __LINE__, __func__, sem); abort(); } +#endif // SEM_USE_PTHREAD + p->valid = 1; *sem = p; @@ -58,7 +165,30 @@ int tsem_wait(tsem_t *sem) { fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", basename(__FILE__), __LINE__, __func__, sem); abort(); } +#ifdef SEM_USE_PTHREAD + if (pthread_mutex_lock(&p->lock)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } + p->val -= 1; + if (p->val < 0) { + if (pthread_cond_wait(&p->cond, &p->lock)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } + } + if (pthread_mutex_unlock(&p->lock)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } + return 0; +#elif defined(SEM_USE_POSIX) + return sem_wait(p->sem); +#elif defined(SEM_USE_SEM) + return semaphore_wait(p->sem); +#else // SEM_USE_PTHREAD return dispatch_semaphore_wait(p->sem, DISPATCH_TIME_FOREVER); +#endif // SEM_USE_PTHREAD } int tsem_post(tsem_t *sem) { @@ -71,21 +201,76 @@ int tsem_post(tsem_t *sem) { fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", basename(__FILE__), __LINE__, __func__, sem); abort(); } +#ifdef SEM_USE_PTHREAD + if (pthread_mutex_lock(&p->lock)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } + p->val += 1; + if (p->val <= 0) { + if (pthread_cond_signal(&p->cond)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } + } + if (pthread_mutex_unlock(&p->lock)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } + return 0; +#elif defined(SEM_USE_POSIX) + return sem_post(p->sem); +#elif defined(SEM_USE_SEM) + return semaphore_signal(p->sem); +#else // SEM_USE_PTHREAD return dispatch_semaphore_signal(p->sem); +#endif // SEM_USE_PTHREAD } int tsem_destroy(tsem_t *sem) { - fprintf(stderr, "==%s[%d]%s():[%p]==destroying\n", basename(__FILE__), __LINE__, __func__, sem); + // fprintf(stderr, "==%s[%d]%s():[%p]==destroying\n", basename(__FILE__), __LINE__, __func__, sem); if (!*sem) { - fprintf(stderr, "==%s[%d]%s():[%p]==not initialized\n", basename(__FILE__), __LINE__, __func__, sem); - abort(); + // fprintf(stderr, "==%s[%d]%s():[%p]==not initialized\n", basename(__FILE__), __LINE__, __func__, sem); + // abort(); return 0; } struct tsem_s *p = *sem; if (!p->valid) { - fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", basename(__FILE__), __LINE__, __func__, sem); + // fprintf(stderr, "==%s[%d]%s():[%p]==already destroyed\n", basename(__FILE__), __LINE__, __func__, sem); + // abort(); + return 0; + } +#ifdef SEM_USE_PTHREAD + if (pthread_mutex_lock(&p->lock)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } + p->valid = 0; + if (pthread_cond_destroy(&p->cond)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } + if (pthread_mutex_unlock(&p->lock)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } + if (pthread_mutex_destroy(&p->lock)) { + fprintf(stderr, "==%s[%d]%s():[%p]==internal logic error\n", basename(__FILE__), __LINE__, __func__, sem); + abort(); + } +#elif defined(SEM_USE_POSIX) + char name[NAME_MAX-4]; + snprintf(name, sizeof(name), "/t%ld", p->id); + int r = sem_unlink(name); + if (r) { + int e = errno; + fprintf(stderr, "==%s[%d]%s():[%p]==unlink failed[%d]%s\n", basename(__FILE__), __LINE__, __func__, sem, e, strerror(e)); abort(); } +#elif defined(SEM_USE_SEM) + semaphore_destroy(sem_port, p->sem); +#else // SEM_USE_PTHREAD +#endif // SEM_USE_PTHREAD p->valid = 0; free(p); diff --git a/src/os/src/darwin/darwinTimer.c b/src/os/src/darwin/darwinTimer.c index 121e2e9427a7e313db9f718883d5f21ff39c3dad..ee1becc91af5237296bb2a562982b33cc5c78f10 100644 --- a/src/os/src/darwin/darwinTimer.c +++ b/src/os/src/darwin/darwinTimer.c @@ -13,9 +13,82 @@ * along with this program. If not, see . */ +// fail-fast or let-it-crash philosophy +// https://en.wikipedia.org/wiki/Fail-fast +// https://stackoverflow.com/questions/4393197/erlangs-let-it-crash-philosophy-applicable-elsewhere +// experimentally, we follow log-and-crash here + #define _DEFAULT_SOURCE #include "os.h" +#if 1 +#include + +static void (*timer_callback)(int); +static int timer_ms = 0; +static pthread_t timer_thread; +static int timer_kq = -1; +static volatile int timer_stop = 0; + +static void* timer_routine(void *arg) { + (void)arg; + + int r = 0; + struct timespec to = {0}; + to.tv_sec = timer_ms / 1000; + to.tv_nsec = (timer_ms % 1000) * 1000000; + while (!timer_stop) { + struct kevent64_s kev[10] = {0}; + r = kevent64(timer_kq, NULL, 0, kev, sizeof(kev)/sizeof(kev[0]), 0, &to); + if (r!=0) { + fprintf(stderr, "==%s[%d]%s()==kevent64 failed\n", basename(__FILE__), __LINE__, __func__); + abort(); + } + timer_callback(SIGALRM); // just mock + } + + return NULL; +} + +int taosInitTimer(void (*callback)(int), int ms) { + int r = 0; + timer_ms = ms; + timer_callback = callback; + + timer_kq = kqueue(); + if (timer_kq==-1) { + fprintf(stderr, "==%s[%d]%s()==failed to create timer kq\n", basename(__FILE__), __LINE__, __func__); + // since no caller of this func checks the return value for the moment + abort(); + } + + r = pthread_create(&timer_thread, NULL, timer_routine, NULL); + if (r) { + fprintf(stderr, "==%s[%d]%s()==failed to create timer thread\n", basename(__FILE__), __LINE__, __func__); + // since no caller of this func checks the return value for the moment + abort(); + } + return 0; +} + +void taosUninitTimer() { + int r = 0; + timer_stop = 1; + r = pthread_join(timer_thread, NULL); + if (r) { + fprintf(stderr, "==%s[%d]%s()==failed to join timer thread\n", basename(__FILE__), __LINE__, __func__); + // since no caller of this func checks the return value for the moment + abort(); + } + close(timer_kq); + timer_kq = -1; +} + +void taos_block_sigalrm(void) { + // we don't know if there's any specific API for SIGALRM to deliver to specific thread + // this implementation relies on kqueue rather than SIGALRM +} +#else int taosInitTimer(void (*callback)(int), int ms) { signal(SIGALRM, callback); @@ -46,4 +119,5 @@ void taos_block_sigalrm(void) { already_set = 1; } } +#endif diff --git a/src/os/src/darwin/eok.c b/src/os/src/darwin/eok.c index 9c6a263a40c593cd2484faac8381eb529303b1f7..f731aaf330cb8618aa8a2dc51401ac45ecdc5440 100644 --- a/src/os/src/darwin/eok.c +++ b/src/os/src/darwin/eok.c @@ -13,13 +13,18 @@ * along with this program. If not, see . */ +// fail-fast or let-it-crash philosophy +// https://en.wikipedia.org/wiki/Fail-fast +// https://stackoverflow.com/questions/4393197/erlangs-let-it-crash-philosophy-applicable-elsewhere +// experimentally, we follow log-and-crash here + #include "eok.h" #include "os.h" #include -#define LET_IT_BE +// #define BALANCE_CHECK_WHEN_CLOSE #ifdef ENABLE_LOG #define D(fmt, ...) fprintf(stderr, "%s[%d]%s(): " fmt "\n", basename(__FILE__), __LINE__, __func__, ##__VA_ARGS__) @@ -99,16 +104,16 @@ struct eok_event_s { typedef struct eoks_s eoks_t; struct eoks_s { pthread_mutex_t lock; - ep_over_kq_t **eoks; // note: this memory leaks when process terminates - int neoks; // we can add an extra api to let user clean - ep_over_kq_t *eoks_free; // currently, we just keep it simple stupid + ep_over_kq_t **eoks; // note: this memory leaks when process terminates + int neoks; // we can add an extra api to let user clean + ep_over_kq_t *eoks_free_list; // currently, we just keep it simple stupid }; static eoks_t eoks = { - .lock = PTHREAD_MUTEX_INITIALIZER, - .eoks = NULL, - .neoks = 0, - .eoks_free = NULL, + .lock = PTHREAD_MUTEX_INITIALIZER, + .eoks = NULL, + .neoks = 0, + .eoks_free_list = NULL, }; #ifdef ENABLE_LOG @@ -760,9 +765,9 @@ static ep_over_kq_t* eoks_alloc(void) { A(0==pthread_mutex_lock(&eoks.lock), ""); do { - if (eoks.eoks_free) { - eok = eoks.eoks_free; - eoks.eoks_free = eok->next; + if (eoks.eoks_free_list) { + eok = eoks.eoks_free_list; + eoks.eoks_free_list = eok->next; A(eoks.eoks, "internal logic error"); A(eok->idx>=0 && eok->idxidx)==NULL, "internal logic error"); @@ -820,10 +825,12 @@ static void eoks_free(ep_over_kq_t *eok) { A(eok->waiting==0, "internal logic error"); eok_event_t *ev = eok->evs_head; + int sv_closed = 0; while (ev) { eok_event_t *next = ev->next; if (ev->fd==eok->sv[0]) { // fd is critical system resource + A(sv_closed==0, "internal logic error"); close(eok->sv[0]); eok->sv[0] = -1; close(eok->sv[1]); @@ -832,11 +839,11 @@ static void eoks_free(ep_over_kq_t *eok) { } else { // user forget calling epoll_ctl(EPOLL_CTL_DEL) before calling epoll_close/close? // calling close(ev->fd) here smells really bad -#ifdef LET_IT_BE +#ifndef BALANCE_CHECK_WHEN_CLOSE // we just let it be and reclaim ev eok_free_ev(eok, ev); #else - // panic otherwise, if LET_IT_BE not defined + // panic otherwise, if BALANCE_CHECK_WHEN_CLOSE is defined A(eok->evs_head==NULL && eok->evs_tail==NULL && eok->evs_count==0, "epfd[%d] fd[%d]: internal logic error: have you epoll_ctl(EPOLL_CTL_DEL) everything before calling epoll_close?", eok->idx, ev->fd); @@ -861,8 +868,8 @@ static void eoks_free(ep_over_kq_t *eok) { close(eok->kq); eok->kq = -1; } - eok->next = eoks.eoks_free; - eoks.eoks_free = eok; + eok->next = eoks.eoks_free_list; + eoks.eoks_free_list = eok; *(eoks.eoks + eok->idx) = NULL; } while (0); A(0==pthread_mutex_unlock(&eoks.lock), "");