btif/src/btif_sock_thread.c

   1 /******************************************************************************
   2  *
   3  *  Copyright (C) 2009-2012 Broadcom Corporation
   4  *
   5  *  Licensed under the Apache License, Version 2.0 (the "License");
   6  *  you may not use this file except in compliance with the License.
   7  *  You may obtain a copy of the License at:
   8  *
   9  *  http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  *  Unless required by applicable law or agreed to in writing, software
  12  *  distributed under the License is distributed on an "AS IS" BASIS,
  13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  *  See the License for the specific language governing permissions and
  15  *  limitations under the License.
  16  *
  17  ******************************************************************************/
  18
  19 /************************************************************************************
  20  *
  21  *  Filename:      btif_sock_thread.c
  22  *
  23  *  Description:   socket select thread
  24  *
  25  *
  26  ***********************************************************************************/
  27
  28 #include <hardware/bluetooth.h>
  29 #include <hardware/bt_sock.h>
  30
  31 //bta_jv_co_rfc_data
  32 #include <stdio.h>
  33 #include <stdlib.h>
  34 #include <errno.h>
  35 #include <features.h>
  36 #include <string.h>
  37 #include <sys/types.h>
  38 #include <sys/socket.h>
  39 #include <sys/un.h>
  40 #include <time.h>
  41 #include <fcntl.h>
  42 #include <unistd.h>
  43 #include <signal.h>
  44 #include <pthread.h>
  45 #include <ctype.h>
  46
  47 #include <sys/select.h>
  48 #include <sys/poll.h>
  49 #include <cutils/sockets.h>
  50 #include <alloca.h>
  51
  52 #define LOG_TAG "bt_btif_sock"
  53 #include "btif_common.h"
  54 #include "btif_util.h"
  55
  56
  57 #include "bta_api.h"
  58 #include "btif_sock.h"
  59 #include "btif_sock_thread.h"
  60 #include "btif_sock_util.h"
  61
  62 #define asrt(s) if(!(s)) APPL_TRACE_ERROR("## %s assert %s failed at line:%d ##",__FUNCTION__, #s, __LINE__)
  63 #define print_events(events) do { \
  64     APPL_TRACE_DEBUG("print poll event:%x", events); \
  65     if (events & POLLIN) APPL_TRACE_DEBUG(  "   POLLIN "); \
  66     if (events & POLLPRI) APPL_TRACE_DEBUG( "   POLLPRI "); \
  67     if (events & POLLOUT) APPL_TRACE_DEBUG( "   POLLOUT "); \
  68     if (events & POLLERR) APPL_TRACE_DEBUG( "   POLLERR "); \
  69     if (events & POLLHUP) APPL_TRACE_DEBUG( "   POLLHUP "); \
  70     if (events & POLLNVAL) APPL_TRACE_DEBUG("   POLLNVAL "); \
  71     if (events & POLLRDHUP) APPL_TRACE_DEBUG("   POLLRDHUP"); \
  72     } while(0)
  73
  74 #define MAX_THREAD 8
  75 #define MAX_POLL 64
  76 #define POLL_EXCEPTION_EVENTS (POLLHUP | POLLRDHUP | POLLERR | POLLNVAL)
  77 #define IS_EXCEPTION(e) ((e) & POLL_EXCEPTION_EVENTS)
  78 #define IS_READ(e) ((e) & POLLIN)
  79 #define IS_WRITE(e) ((e) & POLLOUT)
  80 /*cmd executes in socket poll thread */
  81 #define CMD_WAKEUP       1
  82 #define CMD_EXIT         2
  83 #define CMD_ADD_FD       3
  84 #define CMD_REMOVE_FD    4
  85 #define CMD_USER_PRIVATE 5
  86
  87 typedef struct {
  88     struct pollfd pfd;
  89     uint32_t user_id;
  90     int type;
  91     int flags;
  92 } poll_slot_t;
  93 typedef struct {
  94     int cmd_fdr, cmd_fdw;
  95     int poll_count;
  96     poll_slot_t ps[MAX_POLL];
  97     int psi[MAX_POLL]; //index of poll slot
  98     volatile pthread_t thread_id;
  99     btsock_signaled_cb callback;
 100     btsock_cmd_cb cmd_callback;
 101     int used;
 102 } thread_slot_t;
 103 static thread_slot_t ts[MAX_THREAD];
 104
 105
 106
 107 static void *sock_poll_thread(void *arg);
 108 static inline void close_cmd_fd(int h);
 109
 110 static inline void add_poll(int h, int fd, int type, int flags, uint32_t user_id);
 111
 112 static pthread_mutex_t thread_slot_lock = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
 113
 114 static inline void set_socket_blocking(int s, int blocking)
 115 {
 116     int opts;
 117     opts = fcntl(s, F_GETFL);
 118     if (opts<0) APPL_TRACE_ERROR("set blocking (%s)", strerror(errno));
 119     if(blocking)
 120         opts &= ~O_NONBLOCK;
 121     else opts |= O_NONBLOCK;
 122     if (fcntl(s, F_SETFL, opts) < 0)
 123         APPL_TRACE_ERROR("set blocking (%s)", strerror(errno));
 124 }
 125
 126 static inline int create_server_socket(const char* name)
 127 {
 128     int s = socket(AF_LOCAL, SOCK_STREAM, 0);
 129     if(s < 0)
 130         return -1;
 131     APPL_TRACE_DEBUG("covert name to android abstract name:%s", name);
 132     if(socket_local_server_bind(s, name, ANDROID_SOCKET_NAMESPACE_ABSTRACT) >= 0)
 133     {
 134         if(listen(s, 5) == 0)
 135         {
 136             APPL_TRACE_DEBUG("listen to local socket:%s, fd:%d", name, s);
 137             return s;
 138         }
 139         else APPL_TRACE_ERROR("listen to local socket:%s, fd:%d failed, errno:%d", name, s, errno);
 140     }
 141     else APPL_TRACE_ERROR("create local socket:%s fd:%d, failed, errno:%d", name, s, errno);
 142     close(s);
 143     return -1;
 144 }
 145 static inline int connect_server_socket(const char* name)
 146 {
 147     int s = socket(AF_LOCAL, SOCK_STREAM, 0);
 148     if(s < 0)
 149         return -1;
 150     set_socket_blocking(s, TRUE);
 151     if(socket_local_client_connect(s, name, ANDROID_SOCKET_NAMESPACE_ABSTRACT, SOCK_STREAM) >= 0)
 152     {
 153         APPL_TRACE_DEBUG("connected to local socket:%s, fd:%d", name, s);
 154         return s;
 155     }
 156     else APPL_TRACE_ERROR("connect to local socket:%s, fd:%d failed, errno:%d", name, s, errno);
 157     close(s);
 158     return -1;
 159 }
 160 static inline int accept_server_socket(int s)
 161 {
 162     struct sockaddr_un client_address;
 163     socklen_t clen;
 164     int fd = accept(s, (struct sockaddr*)&client_address, &clen);
 165     APPL_TRACE_DEBUG("accepted fd:%d for server fd:%d", fd, s);
 166     return fd;
 167 }
 168
 169 static inline int create_thread(void *(*start_routine)(void *), void * arg,
 170                                 pthread_t * thread_id)
 171 {
 172     pthread_attr_t thread_attr;
 173     pthread_attr_init(&thread_attr);
 174     pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_JOINABLE);
 175     int policy;
 176     int min_pri=0;
 177         int ret = -1;
 178     struct sched_param param;
 179
 180     if ((ret = pthread_create(thread_id, &thread_attr, start_routine, arg))!=0 )
 181     {
 182         APPL_TRACE_ERROR("pthread_create : %s", strerror(errno));
 183         return ret;
 184     }
 185     /* We need to lower the priority of this thread to ensure the stack gets
 186      * priority over transfer to a socket */
 187     pthread_getschedparam(*thread_id, &policy, &param);
 188     min_pri = sched_get_priority_min(policy);
 189     if (param.sched_priority > min_pri) {
 190         param.sched_priority -= 1;
 191     }
 192     pthread_setschedparam(*thread_id, policy, &param);
 193     return ret;
 194 }
 195 static void init_poll(int cmd_fd);
 196 static int alloc_thread_slot()
 197 {
 198     int i;
 199     //revserd order to save guard uninitialized access to 0 index
 200     for(i = MAX_THREAD - 1; i >=0; i--)
 201     {
 202         APPL_TRACE_DEBUG("ts[%d].used:%d", i, ts[i].used);
 203         if(!ts[i].used)
 204         {
 205             ts[i].used = 1;
 206             return i;
 207         }
 208     }
 209     APPL_TRACE_ERROR("execeeded max thread count");
 210     return -1;
 211 }
 212 static void free_thread_slot(int h)
 213 {
 214     if(0 <= h && h < MAX_THREAD)
 215     {
 216         close_cmd_fd(h);
 217         ts[h].used = 0;
 218     }
 219     else APPL_TRACE_ERROR("invalid thread handle:%d", h);
 220 }
 221 int btsock_thread_init()
 222 {
 223     static int initialized;
 224     APPL_TRACE_DEBUG("in initialized:%d", initialized);
 225     if(!initialized)
 226     {
 227         initialized = 1;
 228         int h;
 229         for(h = 0; h < MAX_THREAD; h++)
 230         {
 231             ts[h].cmd_fdr = ts[h].cmd_fdw = -1;
 232             ts[h].used = 0;
 233             ts[h].thread_id = -1;
 234             ts[h].poll_count = 0;
 235             ts[h].callback = NULL;
 236             ts[h].cmd_callback = NULL;
 237         }
 238     }
 239     return TRUE;
 240 }
 241 int btsock_thread_create(btsock_signaled_cb callback, btsock_cmd_cb cmd_callback)
 242 {
 243     asrt(callback || cmd_callback);
 244     pthread_mutex_lock(&thread_slot_lock);
 245     int h = alloc_thread_slot();
 246     pthread_mutex_unlock(&thread_slot_lock);
 247     APPL_TRACE_DEBUG("alloc_thread_slot ret:%d", h);
 248     if(h >= 0)
 249     {
 250         init_poll(h);
 251         pthread_t thread;
 252         int status = create_thread(sock_poll_thread, (void*)(uintptr_t)h, &thread);
 253         if (status)
 254         {
 255             APPL_TRACE_ERROR("create_thread failed: %s", strerror(status));
 256             free_thread_slot(h);
 257             return -1;
 258         }
 259
 260         ts[h].thread_id = thread;
 261         APPL_TRACE_DEBUG("h:%d, thread id:%d", h, ts[h].thread_id);
 262         ts[h].callback = callback;
 263         ts[h].cmd_callback = cmd_callback;
 264     }
 265     return h;
 266 }
 267
 268 /* create dummy socket pair used to wake up select loop */
 269 static inline void init_cmd_fd(int h)
 270 {
 271     asrt(ts[h].cmd_fdr == -1 && ts[h].cmd_fdw == -1);
 272     if(socketpair(AF_UNIX, SOCK_STREAM, 0, &ts[h].cmd_fdr) < 0)
 273     {
 274         APPL_TRACE_ERROR("socketpair failed: %s", strerror(errno));
 275         return;
 276     }
 277     APPL_TRACE_DEBUG("h:%d, cmd_fdr:%d, cmd_fdw:%d", h, ts[h].cmd_fdr, ts[h].cmd_fdw);
 278     //add the cmd fd for read & write
 279     add_poll(h, ts[h].cmd_fdr, 0, SOCK_THREAD_FD_RD, 0);
 280 }
 281 static inline void close_cmd_fd(int h)
 282 {
 283     if(ts[h].cmd_fdr != -1)
 284     {
 285         close(ts[h].cmd_fdr);
 286         ts[h].cmd_fdr = -1;
 287     }
 288     if(ts[h].cmd_fdw != -1)
 289     {
 290         close(ts[h].cmd_fdw);
 291         ts[h].cmd_fdw = -1;
 292     }
 293 }
 294 typedef struct
 295 {
 296     int id;
 297     int fd;
 298     int type;
 299     int flags;
 300     uint32_t user_id;
 301 } sock_cmd_t;
 302 int btsock_thread_add_fd(int h, int fd, int type, int flags, uint32_t user_id)
 303 {
 304     if(h < 0 || h >= MAX_THREAD)
 305     {
 306         APPL_TRACE_ERROR("invalid bt thread handle:%d", h);
 307         return FALSE;
 308     }
 309     if(ts[h].cmd_fdw == -1)
 310     {
 311         APPL_TRACE_ERROR("cmd socket is not created. socket thread may not initialized");
 312         return FALSE;
 313     }
 314     if(flags & SOCK_THREAD_ADD_FD_SYNC)
 315     {
 316         //must executed in socket poll thread
 317         if(ts[h].thread_id == pthread_self())
 318         {
 319             //cleanup one-time flags
 320             flags &= ~SOCK_THREAD_ADD_FD_SYNC;
 321             add_poll(h, fd, type, flags, user_id);
 322             return TRUE;
 323         }
 324         APPL_TRACE_DEBUG("THREAD_ADD_FD_SYNC is not called in poll thread, fallback to async");
 325     }
 326     sock_cmd_t cmd = {CMD_ADD_FD, fd, type, flags, user_id};
 327     APPL_TRACE_DEBUG("adding fd:%d, flags:0x%x", fd, flags);
 328     return send(ts[h].cmd_fdw, &cmd, sizeof(cmd), 0) == sizeof(cmd);
 329 }
 330
 331 bool btsock_thread_remove_fd_and_close(int thread_handle, int fd)
 332 {
 333     if (thread_handle < 0 || thread_handle >= MAX_THREAD)
 334     {
 335         APPL_TRACE_ERROR("%s invalid thread handle: %d", __func__, thread_handle);
 336         return false;
 337     }
 338     if (fd == -1)
 339     {
 340         APPL_TRACE_ERROR("%s invalid file descriptor.", __func__);
 341         return false;
 342     }
 343
 344     sock_cmd_t cmd = {CMD_REMOVE_FD, fd, 0, 0, 0};
 345     return send(ts[thread_handle].cmd_fdw, &cmd, sizeof(cmd), 0) == sizeof(cmd);
 346 }
 347
 348 int btsock_thread_post_cmd(int h, int type, const unsigned char* data, int size, uint32_t user_id)
 349 {
 350     if(h < 0 || h >= MAX_THREAD)
 351     {
 352         APPL_TRACE_ERROR("invalid bt thread handle:%d", h);
 353         return FALSE;
 354     }
 355     if(ts[h].cmd_fdw == -1)
 356     {
 357         APPL_TRACE_ERROR("cmd socket is not created. socket thread may not initialized");
 358         return FALSE;
 359     }
 360     sock_cmd_t cmd = {CMD_USER_PRIVATE, 0, type, size, user_id};
 361     APPL_TRACE_DEBUG("post cmd type:%d, size:%d, h:%d, ", type, size, h);
 362     sock_cmd_t* cmd_send = &cmd;
 363     int size_send = sizeof(cmd);
 364     if(data && size)
 365     {
 366         size_send = sizeof(cmd) + size;
 367         cmd_send = (sock_cmd_t*)alloca(size_send);
 368         if(cmd_send)
 369         {
 370             *cmd_send = cmd;
 371             memcpy(cmd_send + 1, data, size);
 372         }
 373         else
 374         {
 375             APPL_TRACE_ERROR("alloca failed at h:%d, cmd type:%d, size:%d", h, type, size_send);
 376             return FALSE;
 377         }
 378     }
 379     return send(ts[h].cmd_fdw, cmd_send, size_send, 0) == size_send;
 380 }
 381 int btsock_thread_wakeup(int h)
 382 {
 383     if(h < 0 || h >= MAX_THREAD)
 384     {
 385         APPL_TRACE_ERROR("invalid bt thread handle:%d", h);
 386         return FALSE;
 387     }
 388     if(ts[h].cmd_fdw == -1)
 389     {
 390         APPL_TRACE_ERROR("thread handle:%d, cmd socket is not created", h);
 391         return FALSE;
 392     }
 393     sock_cmd_t cmd = {CMD_WAKEUP, 0, 0, 0, 0};
 394     return send(ts[h].cmd_fdw, &cmd, sizeof(cmd), 0) == sizeof(cmd);
 395 }
 396 int btsock_thread_exit(int h)
 397 {
 398     if(h < 0 || h >= MAX_THREAD)
 399     {
 400         APPL_TRACE_ERROR("invalid bt thread handle:%d", h);
 401         return FALSE;
 402     }
 403     if(ts[h].cmd_fdw == -1)
 404     {
 405         APPL_TRACE_ERROR("cmd socket is not created");
 406         return FALSE;
 407     }
 408     sock_cmd_t cmd = {CMD_EXIT, 0, 0, 0, 0};
 409     if(send(ts[h].cmd_fdw, &cmd, sizeof(cmd), 0) == sizeof(cmd))
 410     {
 411         pthread_join(ts[h].thread_id, 0);
 412         pthread_mutex_lock(&thread_slot_lock);
 413         free_thread_slot(h);
 414         pthread_mutex_unlock(&thread_slot_lock);
 415         return TRUE;
 416     }
 417     return FALSE;
 418 }
 419 static void init_poll(int h)
 420 {
 421     int i;
 422     ts[h].poll_count = 0;
 423     ts[h].thread_id = -1;
 424     ts[h].callback = NULL;
 425     ts[h].cmd_callback = NULL;
 426     for(i = 0; i < MAX_POLL; i++)
 427     {
 428         ts[h].ps[i].pfd.fd = -1;
 429         ts[h].psi[i] = -1;
 430     }
 431     init_cmd_fd(h);
 432 }
 433 static inline unsigned int flags2pevents(int flags)
 434 {
 435     unsigned int pevents = 0;
 436     if(flags & SOCK_THREAD_FD_WR)
 437         pevents |= POLLOUT;
 438     if(flags & SOCK_THREAD_FD_RD)
 439         pevents |= POLLIN;
 440     pevents |= POLL_EXCEPTION_EVENTS;
 441     return pevents;
 442 }
 443
 444 static inline void set_poll(poll_slot_t* ps, int fd, int type, int flags, uint32_t user_id)
 445 {
 446     ps->pfd.fd = fd;
 447     ps->user_id = user_id;
 448     if(ps->type != 0 && ps->type != type)
 449         APPL_TRACE_ERROR("poll socket type should not changed! type was:%d, type now:%d", ps->type, type);
 450     ps->type = type;
 451     ps->flags = flags;
 452     ps->pfd.events = flags2pevents(flags);
 453     ps->pfd.revents = 0;
 454 }
 455 static inline void add_poll(int h, int fd, int type, int flags, uint32_t user_id)
 456 {
 457     asrt(fd != -1);
 458     int i;
 459     int empty = -1;
 460     poll_slot_t* ps = ts[h].ps;
 461
 462     for(i = 0; i < MAX_POLL; i++)
 463     {
 464         if(ps[i].pfd.fd == fd)
 465         {
 466             asrt(ts[h].poll_count < MAX_POLL);
 467
 468             set_poll(&ps[i], fd, type, flags | ps[i].flags, user_id);
 469             return;
 470         }
 471         else if(empty < 0 && ps[i].pfd.fd == -1)
 472             empty = i;
 473     }
 474     if(empty >= 0)
 475     {
 476         asrt(ts[h].poll_count < MAX_POLL);
 477         set_poll(&ps[empty], fd, type, flags, user_id);
 478         ++ts[h].poll_count;
 479         return;
 480     }
 481     APPL_TRACE_ERROR("exceeded max poll slot:%d!", MAX_POLL);
 482 }
 483 static inline void remove_poll(int h, poll_slot_t* ps, int flags)
 484 {
 485     if(flags == ps->flags)
 486     {
 487         //all monitored events signaled. To remove it, just clear the slot
 488         --ts[h].poll_count;
 489         memset(ps, 0, sizeof(*ps));
 490         ps->pfd.fd = -1;
 491     }
 492     else
 493     {
 494         //one read or one write monitor event signaled, removed the accordding bit
 495         ps->flags &= ~flags;
 496         //update the poll events mask
 497         ps->pfd.events = flags2pevents(ps->flags);
 498     }
 499 }
 500 static int process_cmd_sock(int h)
 501 {
 502     sock_cmd_t cmd = {-1, 0, 0, 0, 0};
 503     int fd = ts[h].cmd_fdr;
 504     if(recv(fd, &cmd, sizeof(cmd), MSG_WAITALL) != sizeof(cmd))
 505     {
 506         APPL_TRACE_ERROR("recv cmd errno:%d", errno);
 507         return FALSE;
 508     }
 509     APPL_TRACE_DEBUG("cmd.id:%d", cmd.id);
 510     switch(cmd.id)
 511     {
 512         case CMD_ADD_FD:
 513             add_poll(h, cmd.fd, cmd.type, cmd.flags, cmd.user_id);
 514             break;
 515         case CMD_REMOVE_FD:
 516             for (int i = 1; i < MAX_POLL; ++i)
 517             {
 518                 poll_slot_t *poll_slot = &ts[h].ps[i];
 519                 if (poll_slot->pfd.fd == cmd.fd)
 520                 {
 521                     remove_poll(h, poll_slot, poll_slot->flags);
 522                     break;
 523                 }
 524             }
 525             close(cmd.fd);
 526             break;
 527         case CMD_WAKEUP:
 528             break;
 529         case CMD_USER_PRIVATE:
 530             asrt(ts[h].cmd_callback);
 531             if(ts[h].cmd_callback)
 532                 ts[h].cmd_callback(fd, cmd.type, cmd.flags, cmd.user_id);
 533             break;
 534         case CMD_EXIT:
 535             return FALSE;
 536         default:
 537             APPL_TRACE_DEBUG("unknown cmd: %d", cmd.id);
 538              break;
 539     }
 540     return TRUE;
 541 }
 542 static void process_data_sock(int h, struct pollfd *pfds, int count)
 543 {
 544     asrt(count <= ts[h].poll_count);
 545     int i;
 546     for( i= 1; i < ts[h].poll_count; i++)
 547     {
 548         if(pfds[i].revents)
 549         {
 550             int ps_i = ts[h].psi[i];
 551             asrt(pfds[i].fd == ts[h].ps[ps_i].pfd.fd);
 552             uint32_t user_id = ts[h].ps[ps_i].user_id;
 553             int type = ts[h].ps[ps_i].type;
 554             int flags = 0;
 555             print_events(pfds[i].revents);
 556             if(IS_READ(pfds[i].revents))
 557             {
 558                 flags |= SOCK_THREAD_FD_RD;
 559             }
 560             if(IS_WRITE(pfds[i].revents))
 561             {
 562                 flags |= SOCK_THREAD_FD_WR;
 563             }
 564             if(IS_EXCEPTION(pfds[i].revents))
 565             {
 566                 flags |= SOCK_THREAD_FD_EXCEPTION;
 567                 //remove the whole slot not flags
 568                 remove_poll(h, &ts[h].ps[ps_i], ts[h].ps[ps_i].flags);
 569             }
 570             else if(flags)
 571                  remove_poll(h, &ts[h].ps[ps_i], flags); //remove the monitor flags that already processed
 572             if(flags)
 573                 ts[h].callback(pfds[i].fd, type, flags, user_id);
 574         }
 575     }
 576 }
 577
 578 static void prepare_poll_fds(int h, struct pollfd* pfds)
 579 {
 580     int count = 0;
 581     int ps_i = 0;
 582     int pfd_i = 0;
 583     asrt(ts[h].poll_count <= MAX_POLL);
 584     memset(pfds, 0, sizeof(pfds[0])*ts[h].poll_count);
 585     while(count < ts[h].poll_count)
 586     {
 587         if(ps_i >= MAX_POLL)
 588         {
 589             APPL_TRACE_ERROR("exceed max poll range, ps_i:%d, MAX_POLL:%d, count:%d, ts[h].poll_count:%d",
 590                     ps_i, MAX_POLL, count, ts[h].poll_count);
 591             return;
 592         }
 593         if(ts[h].ps[ps_i].pfd.fd >= 0)
 594         {
 595             pfds[pfd_i] =  ts[h].ps[ps_i].pfd;
 596             ts[h].psi[pfd_i] = ps_i;
 597             count++;
 598             pfd_i++;
 599         }
 600         ps_i++;
 601     }
 602 }
 603 static void *sock_poll_thread(void *arg)
 604 {
 605     struct pollfd pfds[MAX_POLL];
 606     memset(pfds, 0, sizeof(pfds));
 607     int h = (intptr_t)arg;
 608     for(;;)
 609     {
 610         prepare_poll_fds(h, pfds);
 611         int ret = poll(pfds, ts[h].poll_count, -1);
 612         if(ret == -1)
 613         {
 614             APPL_TRACE_ERROR("poll ret -1, exit the thread, errno:%d, err:%s", errno, strerror(errno));
 615             break;
 616         }
 617         if(ret != 0)
 618         {
 619             int need_process_data_fd = TRUE;
 620             if(pfds[0].revents) //cmd fd always is the first one
 621             {
 622                 asrt(pfds[0].fd == ts[h].cmd_fdr);
 623                 if(!process_cmd_sock(h))
 624                 {
 625                     APPL_TRACE_DEBUG("h:%d, process_cmd_sock return false, exit...", h);
 626                     break;
 627                 }
 628                 if(ret == 1)
 629                     need_process_data_fd = FALSE;
 630                 else ret--; //exclude the cmd fd
 631             }
 632             if(need_process_data_fd)
 633                 process_data_sock(h, pfds, ret);
 634         }
 635         else {APPL_TRACE_DEBUG("no data, select ret: %d", ret)};
 636     }
 637     ts[h].thread_id = -1;
 638     APPL_TRACE_DEBUG("socket poll thread exiting, h:%d", h);
 639     return 0;
 640 }