diff --git a/src/kit/shell/src/shellDarwin.c b/src/kit/shell/src/shellDarwin.c index d6aed4401c095fc1db54cabb7d759bd8a0cdb54b..dbec3fdb05761d386d54a8e3e5a0d2d6dcb6bfeb 100644 --- a/src/kit/shell/src/shellDarwin.c +++ b/src/kit/shell/src/shellDarwin.c @@ -350,6 +350,9 @@ void *shellLoopQuery(void *arg) { reset_terminal_mode(); } while (shellRunCommand(con, command) == 0); + tfree(command); + exitShell(); + pthread_cleanup_pop(1); return NULL; diff --git a/src/os/inc/eok.h b/src/os/inc/eok.h index 8892e50c35e4978fb1c10df7fd78f88d54a1a05c..a0fd5b5d8e15ee9722e3ad4b6c29d5a432053174 100644 --- a/src/os/inc/eok.h +++ b/src/os/inc/eok.h @@ -70,5 +70,5 @@ int epoll_close(int epfd); } #endif -#endif // _eok_h_fd274616_996c_400e_9023_ae70be881fa3_ +#endif // _eok_h_fd274616_996c_400e_9023_ae70be881fa3_ diff --git a/src/os/src/darwin/darwinEnv.c b/src/os/src/darwin/darwinEnv.c index 28388f24d24af648a808536ab6834ee488ab417e..da4b32139e837a21a4a5a04c129b5a9a50081c5a 100644 --- a/src/os/src/darwin/darwinEnv.c +++ b/src/os/src/darwin/darwinEnv.c @@ -17,8 +17,6 @@ #include "os.h" #include "tglobal.h" -#include - void osInit() { if (configDir[0] == 0) { strcpy(configDir, "~/TDengine/cfg"); diff --git a/src/os/src/darwin/eok.c b/src/os/src/darwin/eok.c index f83846b73436987d7848a836ee0b153a5dc9393d..3f7137f6a1148bef7f368198a8aa5dbada5ea41d 100644 --- a/src/os/src/darwin/eok.c +++ b/src/os/src/darwin/eok.c @@ -34,20 +34,31 @@ typedef struct eok_event_s eok_event_t; struct ep_over_kq_s { int kq; + + // !!! + // idx in the eoks list + // used as pseudo-file-desciptor + // must be 'closed' with epoll_close int idx; + ep_over_kq_t *next; int sv[2]; // 0 for read, 1 for write + // all registered 'epoll events, key by fd' int evs_count; eok_event_t *evs_head; eok_event_t *evs_tail; eok_event_t *evs_free; + // all kev changes list pending to be processed by kevent64 + // key by tuple (ident,filter), ident === fd in this case struct kevent64_s *kchanges; int nchanges; int ichanges; + // kev eventslist for kevent64 to store active events + // they remain alive among kevent64 calls struct kevent64_s *kevslist; int nevslist; @@ -76,6 +87,7 @@ struct eoks_s { int neoks; ep_over_kq_t *eoks_free; }; + static eoks_t eoks = { .lock = PTHREAD_MUTEX_INITIALIZER, .eoks = NULL, @@ -93,8 +105,9 @@ static const char* op_str(int op) { } static __thread char buf_slots[10][1024] = {0}; -static __thread int buf_slots_linelen = sizeof(buf_slots[0])/sizeof(buf_slots[0][0]); -static __thread int buf_slots_count = sizeof(buf_slots)/(sizeof(buf_slots[0])/sizeof(buf_slots[0][0])); +static __thread int buf_slots_linelen = sizeof(buf_slots[0])/sizeof(buf_slots[0][0]); +static __thread int buf_slots_count = sizeof(buf_slots)/(sizeof(buf_slots[0])/sizeof(buf_slots[0][0])); + static const char* events_str(uint32_t events, int slots) { A(slots>=0 && slots on linux // EPOLLIN = 0x001, // #define EPOLLIN EPOLLIN // EPOLLPRI = 0x002, @@ -164,6 +178,7 @@ static const char* kev_flags_str(uint16_t flags, int slots) { size_t len = buf_slots_linelen; int n = 0; buf[0] = '\0'; + // copied to // #define EV_ADD 0x0001 /* add event to kq (implies enable) */ // #define EV_DELETE 0x0002 /* delete event from kq */ // #define EV_ENABLE 0x0004 /* enable event */ @@ -233,10 +248,7 @@ int epoll_create(int size) { (void)size; int e = 0; ep_over_kq_t *eok = eoks_alloc(); - if (!eok) { - errno = ENOMEM; - return -1; - } + if (!eok) return -1; A(eok->kq==-1, "internal logic error"); A(eok->lock_valid, "internal logic error"); @@ -263,13 +275,14 @@ int epoll_create(int size) { struct epoll_event ev = {0}; ev.events = EPOLLIN; ev.data.ptr = &eok_dummy; + D("epoll_create epfd:[%d]", eok->idx); if (epoll_ctl(eok->idx, EPOLL_CTL_ADD, eok->sv[0], &ev)) { - A(0, "internal logic error"); + e = errno; epoll_close(eok->idx); + errno = e; return -1; } - D("epoll_create epfd:[%d]", eok->idx); return eok->idx; } @@ -309,36 +322,34 @@ int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) { e = ENOENT; break; } + if (op!=EPOLL_CTL_DEL && !event) { + e = EINVAL; + break; + } + // prepare krev/kwev struct kevent64_s krev = {0}; struct kevent64_s kwev = {0}; krev.ident = -1; kwev.ident = -1; uint16_t flags = 0; + // prepare internal eok event eok_event_t ev = {0}; ev.fd = fd; if (event) ev.epev = *event; struct epoll_event *pev = event; switch (op) { case EPOLL_CTL_ADD: { - if (!event) { - e = EINVAL; - break; - } flags = EV_ADD; ev.changed = 1; } break; case EPOLL_CTL_MOD: { - if (!event) { - e = EINVAL; - break; - } flags = EV_ADD; ev.changed = 2; } break; case EPOLL_CTL_DEL: { - D("epoll_ctl adding..."); // event is ignored + // pev points to registered epoll_event pev = &oev->epev; flags = EV_DELETE; ev.changed = 3; @@ -350,6 +361,7 @@ int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) { if (e) break; + // udata will be delayed to be set if (pev->events & (EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP)) { flags |= EV_EOF; EV_SET64(&krev, ev.fd, EVFILT_READ, flags, 0, 0, -1, 0, 0); @@ -358,6 +370,7 @@ int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) { EV_SET64(&kwev, ev.fd, EVFILT_WRITE, flags, 0, 0, -1, 0, 0); } + // refresh registered evlist and changelist in a transaction way if (eok_chgs_refresh(eok, oev, &ev, &krev, &kwev)) { e = errno; A(e, "internal logic error"); @@ -380,11 +393,6 @@ static struct timespec do_timespec_diff(struct timespec *from, struct timespec * int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) { int e = 0; - if (epfd<0 || epfd>=eoks.neoks) { - errno = EBADF; - E("epoll_waiting epfd:[%d], maxevents:[%d], timeout:[%d] failed", epfd, maxevents, timeout); - return -1; - } if (!events) { errno = EINVAL; E("epoll_waiting epfd:[%d], maxevents:[%d], timeout:[%d] failed", epfd, maxevents, timeout); @@ -396,15 +404,6 @@ int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) return -1; } - int r = 0; - - ep_over_kq_t *eok = eoks_find(epfd); - if (!eok) { - errno = EBADF; - E("epoll_waiting epfd:[%d], maxevents:[%d], timeout:[%d] failed", epfd, maxevents, timeout); - return -1; - } - struct timespec abstime = {0}; A(TIME_UTC==timespec_get(&abstime, TIME_UTC), "internal logic error"); @@ -412,7 +411,17 @@ int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) if (timeout<0) timeout = 0; int64_t t = abstime.tv_nsec + timeout * 1000000; abstime.tv_sec += t / 1000000000; - abstime.tv_nsec %= 1000000000; + abstime.tv_nsec = t % 1000000000; + } + + int r = 0; + + ep_over_kq_t *eok = eoks_find(epfd); + if (!eok) { + errno = EBADF; + E("epoll_waiting epfd:[%d], maxevents:[%d], timeout:[%d] failed", epfd, maxevents, timeout); + errno = EBADF; + return -1; } int cnts = 0; @@ -431,43 +440,52 @@ int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) struct timespec now = {0}; A(TIME_UTC==timespec_get(&now, TIME_UTC), "internal logic error"); struct timespec to = do_timespec_diff(&now, &abstime); - struct timespec *pto = NULL; - if (timeout!=-1) { - pto = &to; + struct timespec *pto = &to; + if (timeout==-1) { + pto = NULL; } - eok->changed = 0; - eok->wakenup = 0; - eok->waiting = 1; - + // taking the changelist struct kevent64_s *kchanges = eok->kchanges; int nchanges = eok->nchanges; int ichanges = eok->ichanges; + // let outside world to add changes eok->kchanges = NULL; eok->nchanges = 0; eok->ichanges = 0; + eok->changed = 0; + eok->wakenup = 0; + eok->waiting = 1; + A(0==pthread_mutex_unlock(&eok->lock), ""); if (ichanges>0) { - D("kevent64 changing [%d] changes and waiting...", ichanges); + D("kevent64 epfd[%d] changing [%d] changes and waiting...", eok->idx, ichanges); } errno = 0; r = kevent64(eok->kq, kchanges, ichanges, eventslist, maxevents, 0, pto); e = errno; if (e) { - E("kevent64 waiting done, with r[%d]", r); + E("kevent64 epfd[%d] waiting done, with r[%d]", eok->idx, r); } A(0==pthread_mutex_lock(&eok->lock), ""); eok->waiting = 0; + if (kchanges) { - free(kchanges); - kchanges = NULL; + if (eok->kchanges==NULL) { + // reuse + A(eok->nchanges==0 && eok->ichanges==0, "internal logic error"); + eok->kchanges = kchanges; + eok->nchanges = nchanges; + } else { + free(kchanges); + kchanges = NULL; + } nchanges = 0; ichanges = 0; } - eok->waiting = 0; if (r==0) { A(timeout!=-1, "internal logic error"); } @@ -484,28 +502,37 @@ int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) case EVFILT_READ: { A((ev->epev.events & EPOLLIN), "internal logic errro"); if (ev->epev.data.ptr==&eok_dummy) { + // it's coming from wakeup socket pair char c = '\0'; A(1==recv(kev->ident, &c, 1, 0), "internal logic error"); A(0==memcmp(&c, "1", 1), "internal logic error"); D("wokenup"); + continue; } else { if (ev->changed==3) { D("already requested to delete for fd[%d]", ev->fd); + // TODO: write a unit test for this case // EV_DELETE? continue; } + // converting to epoll_event + // we shall collect all kevents for the uniq fd into one epoll_evnt + // but currently, taos never use EPOLLOUT + // just let it this way for the moment struct epoll_event pev = {0}; pev.data.ptr = ev->epev.data.ptr; pev.events = EPOLLIN; if (kev->flags & EV_EOF) { + // take all these as EOF for the moment pev.events |= (EPOLLHUP | EPOLLERR | EPOLLRDHUP); - D(".........."); } + // rounded to what user care pev.events = pev.events & ev->epev.events; D("events found for fd[%d]: [%04x:%s], which was registered: [%04x:%s], kev_flags: [%04x:%s]", ev->fd, pev.events, events_str(pev.events, 0), ev->epev.events, events_str(ev->epev.events, 1), kev->flags, kev_flags_str(kev->flags, 2)); + // now we get ev and store it events[cnts++] = pev; } } break; @@ -518,6 +545,8 @@ int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) } } if (r>=0) { + // we can safely rule out delete-requested-events from the regitered evlists + // if only changelist are correctly registered eok_event_t *p = eok->evs_head; while (p) { eok_event_t *next = p->next; @@ -528,6 +557,14 @@ int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) p = next; } } + if (cnts==0) { + // if no user-cared-events is up + // we check to see if time is up + A(TIME_UTC==timespec_get(&now, TIME_UTC), "internal logic error"); + to = do_timespec_diff(&now, &abstime); + if (to.tv_sec==0 && to.tv_nsec==0) break; + // time is not up yet, continue loop + } } while (cnts==0); if (cnts>0) { D("kevent64 waiting done with [%d] events", cnts); @@ -540,6 +577,7 @@ int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) return -1; } + // tell user how many events are valid return cnts; } @@ -547,19 +585,20 @@ static struct timespec do_timespec_diff(struct timespec *from, struct timespec * struct timespec delta; delta.tv_sec = to->tv_sec - from->tv_sec; delta.tv_nsec = to->tv_nsec - from->tv_nsec; + // norm and round up while (delta.tv_nsec<0) { delta.tv_sec -= 1; delta.tv_nsec += 1000000000; } + if (delta.tv_sec < 0) { + delta.tv_sec = 0; + delta.tv_nsec = 0; + } return delta; } int epoll_close(int epfd) { D("epoll_closing epfd: [%d]", epfd); - if (epfd<0 || epfd>=eoks.neoks) { - errno = EBADF; - return -1; - } ep_over_kq_t *eok = eoks_find(epfd); if (!eok) { errno = EBADF; @@ -568,9 +607,11 @@ int epoll_close(int epfd) { A(0==pthread_mutex_lock(&eok->lock), ""); do { + // panic if it would be double-closed A(eok->stopping==0, "internal logic error"); eok->stopping = 1; - A(eok->waiting, "internal logic error"); + // panic if epoll_wait is pending + A(eok->waiting==0, "internal logic error"); if (eok->kq!=-1) { close(eok->kq); @@ -608,9 +649,12 @@ static eok_event_t* eok_calloc_ev(ep_over_kq_t *eok) { p = eok->evs_free; eok->evs_free = p->next; p->next = NULL; + A(p->prev==NULL, "internal logic error"); } else { p = (eok_event_t*)calloc(1, sizeof(*p)); if (!p) return NULL; + A(p->next==NULL, "internal logic error"); + A(p->prev==NULL, "internal logic error"); } // dirty link p->prev = eok->evs_tail; @@ -626,9 +670,9 @@ static eok_event_t* eok_calloc_ev(ep_over_kq_t *eok) { static void eok_free_ev(ep_over_kq_t *eok, eok_event_t *ev) { if (ev->prev) ev->prev->next = ev->next; else eok->evs_head = ev->next; - ev->prev = NULL; if (ev->next) ev->next->prev = ev->prev; else eok->evs_tail = ev->prev; + ev->prev = NULL; ev->next = eok->evs_free; eok->evs_free = ev->next; @@ -639,7 +683,7 @@ static void eok_wakeup(ep_over_kq_t *eok) { if (!eok->waiting) return; if (eok->wakenup) return; eok->wakenup = 1; - send(eok->sv[1], "1", 1, 0); + A(1==send(eok->sv[1], "1", 1, 0), ""); } static int eok_chgs_refresh(ep_over_kq_t *eok, eok_event_t *oev, eok_event_t *ev, struct kevent64_s *krev, struct kevent64_s *kwev) { @@ -653,6 +697,8 @@ static int eok_chgs_refresh(ep_over_kq_t *eok, eok_event_t *oev, eok_event_t *ev struct kevent64_s *p = (struct kevent64_s*)realloc(eok->kchanges, sizeof(*p) * (eok->nchanges + 10)); if (!p) { if (ev->changed==1) { + // roll back + A(oev, "internal logic error"); eok_free_ev(eok, oev); } errno = ENOMEM; @@ -662,12 +708,15 @@ static int eok_chgs_refresh(ep_over_kq_t *eok, eok_event_t *oev, eok_event_t *ev eok->nchanges += 10; } + // copy to registered event slot oev->fd = ev->fd; if (ev->changed!=3) { + // if add/mod, copy epoll_event oev->epev = ev->epev; } oev->changed = ev->changed; + // copy to changes list n = 0; if (krev->ident==ev->fd) { krev->udata = (uint64_t)oev; @@ -679,7 +728,7 @@ static int eok_chgs_refresh(ep_over_kq_t *eok, eok_event_t *oev, eok_event_t *ev eok->kchanges[eok->ichanges++] = *kwev; ++n; } - D("add changes[%d] for fd[%d], and changes/registers [%d/%d]", n, ev->fd, eok->ichanges, eok->evs_count); + D("add #changes[%d] for fd[%d], and now #changes/registers [%d/%d]", n, ev->fd, eok->ichanges, eok->evs_count); return 0; } @@ -707,7 +756,10 @@ static ep_over_kq_t* eoks_alloc(void) { } while (0); A(0==pthread_mutex_unlock(&eoks.lock), ""); - if (!eok) return NULL; + if (!eok) { + errno = ENOMEM; + return NULL; + } if (eok->lock_valid) { return eok; } @@ -718,6 +770,7 @@ static ep_over_kq_t* eoks_alloc(void) { } eoks_free(eok); + errno = ENOMEM; return NULL; } @@ -725,16 +778,23 @@ static void eoks_free(ep_over_kq_t *eok) { A(0==pthread_mutex_lock(&eoks.lock), ""); do { A(eok->next==NULL, "internal logic error"); - A(eok->evs_head==NULL, "internal logic error"); + + // leave eok->kchanges as is + A(eok->ichanges==0, "internal logic error"); + A(eok->waiting==0, "internal logic error"); - if (eok->sv[0]!=-1) { + if (eok->evs_count==1) { + A(eok->evs_head && eok->evs_tail && eok->evs_head==eok->evs_tail, "internal logic error"); + A(eok->evs_head->fd==eok->sv[0] && eok->sv[0]!=-1 && eok->sv[1]!=-1, "internal logic error"); + // fd is critical system resource close(eok->sv[0]); eok->sv[0] = -1; - } - if (eok->sv[1]!=-1) { close(eok->sv[1]); eok->sv[1] = -1; + eok_free_ev(eok, eok->evs_head); } + A(eok->evs_head==NULL && eok->evs_tail==NULL && eok->evs_count==0, "internal logic error"); + A(eok->sv[0]==-1 && eok->sv[1]==-1, "internal logic error"); if (eok->kq!=-1) { close(eok->kq); eok->kq = -1; diff --git a/src/os/src/detail/osSocket.c b/src/os/src/detail/osSocket.c index c6cbbe6b83449672ea621366b1e471a84a8a9aaf..659c8510dc91a53f7e011054e8c6879a476b645d 100644 --- a/src/os/src/detail/osSocket.c +++ b/src/os/src/detail/osSocket.c @@ -58,7 +58,6 @@ void taosBlockSIGPIPE() { #ifndef TAOS_OS_FUNC_SOCKET_SETSOCKETOPT int32_t taosSetSockOpt(SOCKET socketfd, int32_t level, int32_t optname, void *optval, int32_t optlen) { - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); return setsockopt(socketfd, level, optname, optval, (socklen_t)optlen); } diff --git a/src/rpc/src/rpcTcp.c b/src/rpc/src/rpcTcp.c index ff2a5882f82d8f3a3da301701d07ca030ab3453b..c65c0db66ba529a9bfe322123a30d0f232d18141 100644 --- a/src/rpc/src/rpcTcp.c +++ b/src/rpc/src/rpcTcp.c @@ -133,7 +133,6 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread } pThreadObj->pollFd = (int64_t)epoll_create(10); // size does not matter - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); if (pThreadObj->pollFd < 0) { tError("%s failed to create TCP epoll", label); code = -1; @@ -294,7 +293,6 @@ void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void * } pThreadObj->pollFd = (SOCKET)epoll_create(10); // size does not matter - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); if (pThreadObj->pollFd < 0) { tError("%s failed to create TCP client epoll", label); free(pThreadObj); @@ -343,10 +341,6 @@ void taosCleanUpTcpClient(void *chandle) { void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port) { SThreadObj * pThreadObj = shandle; - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); - fprintf(stderr, "pThreadObj->ip:%d\n", pThreadObj->ip); - fprintf(stderr, "PF_INET/AF_INET:%d/%d\n", PF_INET, AF_INET); - fprintf(stderr, "ip/port:%x/%d\n", ip, port); SOCKET fd = taosOpenTcpClientSocket(ip, port, pThreadObj->ip); if (fd < 0) return NULL; @@ -358,7 +352,6 @@ void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uin localPort = (uint16_t)ntohs(sin.sin_port); } - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); SFdObj *pFdObj = taosMallocFdObj(pThreadObj, fd); if (pFdObj) { @@ -369,7 +362,6 @@ void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uin pThreadObj->label, thandle, ip, port, localPort, pFdObj, pThreadObj->numOfFds); } else { tError("%s failed to malloc client FdObj(%s)", pThreadObj->label, strerror(errno)); - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); taosCloseSocket(fd); } @@ -492,32 +484,27 @@ static void *taosProcessTcpData(void *param) { if (fdNum < 0) continue; for (int i = 0; i < fdNum; ++i) { - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); pFdObj = events[i].data.ptr; if (events[i].events & EPOLLERR) { tDebug("%s %p FD:%p epoll errors", pThreadObj->label, pFdObj->thandle, pFdObj); - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); taosReportBrokenLink(pFdObj); continue; } if (events[i].events & EPOLLRDHUP) { tDebug("%s %p FD:%p RD hang up", pThreadObj->label, pFdObj->thandle, pFdObj); - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); taosReportBrokenLink(pFdObj); continue; } if (events[i].events & EPOLLHUP) { tDebug("%s %p FD:%p hang up", pThreadObj->label, pFdObj->thandle, pFdObj); - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); taosReportBrokenLink(pFdObj); continue; } if (taosReadTcpData(pFdObj, &recvInfo) < 0) { - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); shutdown(pFdObj->fd, SHUT_WR); continue; } @@ -563,7 +550,6 @@ static SFdObj *taosMallocFdObj(SThreadObj *pThreadObj, SOCKET fd) { event.events = EPOLLIN | EPOLLRDHUP; event.data.ptr = pFdObj; - fprintf(stderr, "==%s[%d]%s()==\n", basename(__FILE__), __LINE__, __func__); if (epoll_ctl(pThreadObj->pollFd, EPOLL_CTL_ADD, fd, &event) < 0) { tfree(pFdObj); terrno = TAOS_SYSTEM_ERROR(errno); diff --git a/src/sync/src/syncTcp.c b/src/sync/src/syncTcp.c index be8585381938150d06c8121555fe005a1504ea9b..1e5761316a5730f3b0dff8d7bfb4582689274eb6 100644 --- a/src/sync/src/syncTcp.c +++ b/src/sync/src/syncTcp.c @@ -136,15 +136,12 @@ void *syncAllocateTcpConn(void *param, int64_t rid, int32_t connFd) { event.events = EPOLLIN | EPOLLRDHUP; event.data.ptr = pConn; - fprintf(stderr, ">>>>>>>>>>>>>>>>>\n"); if (epoll_ctl(pThread->pollFd, EPOLL_CTL_ADD, connFd, &event) < 0) { - fprintf(stderr, "<<<<<<<<<<<<<<<<<\n"); sError("failed to add fd:%d since %s", connFd, strerror(errno)); terrno = TAOS_SYSTEM_ERROR(errno); tfree(pConn); pConn = NULL; } else { - fprintf(stderr, "<<<<<<<<<<<<<<<<<\n"); pThread->numOfFds++; sDebug("%p fd:%d is added to epoll thread, num:%d", pThread, connFd, pThread->numOfFds); } @@ -170,9 +167,7 @@ static void taosProcessBrokenLink(SConnObj *pConn) { (*pInfo->processBrokenLink)(pConn->handleId); pThread->numOfFds--; - fprintf(stderr, "<<<<<<<<<<<<<<<<<\n"); epoll_ctl(pThread->pollFd, EPOLL_CTL_DEL, pConn->fd, NULL); - fprintf(stderr, "<<<<<<<<<<<<<<<<<\n"); sDebug("%p fd:%d is removed from epoll thread, num:%d", pThread, pConn->fd, pThread->numOfFds); taosClose(pConn->fd); tfree(pConn); @@ -287,7 +282,6 @@ static SThreadObj *syncGetTcpThread(SPoolObj *pPool) { pThread->pPool = pPool; pThread->pollFd = epoll_create(10); // size does not matter - fprintf(stderr, "...............\n"); if (pThread->pollFd < 0) { tfree(pThread); return NULL; diff --git a/src/tsdb/src/tsdbMain.c b/src/tsdb/src/tsdbMain.c index 602aea70e3760432f35e3281e9b679a305e03daf..5493188c0933cf8d1c06de1a2ff87ca3ea213596 100644 --- a/src/tsdb/src/tsdbMain.c +++ b/src/tsdb/src/tsdbMain.c @@ -649,13 +649,15 @@ static STsdbRepo *tsdbNewRepo(char *rootDir, STsdbAppH *pAppH, STsdbCfg *pCfg) { #ifdef __APPLE__ pRepo->readyToCommit = sem_open(NULL, O_CREAT, 0644, 1); - if (!pRepo->readyToCommit) { + if (pRepo->readyToCommit==SEM_FAILED) { + code = errno; terrno = TAOS_SYSTEM_ERROR(code); goto _err; } #else code = sem_init(&(pRepo->readyToCommit), 0, 1); if (code != 0) { + code = errno; terrno = TAOS_SYSTEM_ERROR(code); goto _err; } diff --git a/src/util/src/tsocket.c b/src/util/src/tsocket.c index 6b15d3677c2ab23f325eaf04d2e5ad43782dbfc6..9fc10576acc869e31adc69c38a7cd9b5557655e6 100644 --- a/src/util/src/tsocket.c +++ b/src/util/src/tsocket.c @@ -35,7 +35,7 @@ int32_t taosGetFqdn(char *fqdn) { hints.ai_flags = AI_CANONNAME; int32_t ret = getaddrinfo(hostname, NULL, &hints, &result); if (!result) { - uError("failed to get fqdn for hostname <%s>, code:%d, reason:%s", hostname, ret, gai_strerror(ret)); + uError("failed to get fqdn, code:%d, reason:%s", ret, gai_strerror(ret)); return -1; } @@ -342,6 +342,7 @@ int32_t taosKeepTcpAlive(SOCKET sockFd) { } #ifndef __APPLE__ + // all fails on macosx int32_t probes = 3; if (taosSetSockOpt(sockFd, SOL_TCP, TCP_KEEPCNT, (void *)&probes, sizeof(probes)) < 0) { uError("fd:%d setsockopt SO_KEEPCNT failed: %d (%s)", sockFd, errno, strerror(errno));