text: #include "hloop.h"
  #include "hevent.h"
  #include "iowatcher.h"
  
  #include "hdef.h"
  #include "hbase.h"
  #include "hlog.h"
  #include "hmath.h"
  #include "htime.h"
  #include "hsocket.h"
  #include "hthread.h"
  
  #define HLOOP_PAUSE_TIME        10      // ms
  #define HLOOP_MAX_BLOCK_TIME    100     // ms
  #define HLOOP_STAT_TIMEOUT      60000   // ms
  
  #define IO_ARRAY_INIT_SIZE              1024
  #define CUSTOM_EVENT_QUEUE_INIT_SIZE    16
  
  #define SOCKPAIR_WRITE_INDEX    0
  #define SOCKPAIR_READ_INDEX     1
  
  static void __hidle_del(hidle_t* idle);
  static void __htimer_del(htimer_t* timer);
  
  static int timers_compare(const struct heap_node* lhs, const struct heap_node* rhs) {
      return TIMER_ENTRY(lhs)->next_timeout < TIMER_ENTRY(rhs)->next_timeout;
  }
  
  static int hloop_process_idles(hloop_t* loop) {
      int nidles = 0;
      struct list_node* node = loop->idles.next;
      hidle_t* idle = NULL;
      while (node != &loop->idles) {
          idle = IDLE_ENTRY(node);
          node = node->next;
          if (idle->repeat != INFINITE) {
              --idle->repeat;
          }
          if (idle->repeat == 0) {
              // NOTE: Just mark it as destroy and remove from list.
              // Real deletion occurs after hloop_process_pendings.
              __hidle_del(idle);
          }
          EVENT_PENDING(idle);
          ++nidles;
      }
      return nidles;
  }
  
  static int hloop_process_timers(hloop_t* loop) {
      int ntimers = 0;
      htimer_t* timer = NULL;
      uint64_t now_hrtime = hloop_now_hrtime(loop);
      while (loop->timers.root) {
          // NOTE: root of minheap has min timeout.
          timer = TIMER_ENTRY(loop->timers.root);
          if (timer->next_timeout > now_hrtime) {
              break;
          }
          if (timer->repeat != INFINITE) {
              --timer->repeat;
          }
          if (timer->repeat == 0) {
              // NOTE: Just mark it as destroy and remove from heap.
              // Real deletion occurs after hloop_process_pendings.
              __htimer_del(timer);
          }
          else {
              // NOTE: calc next timeout, then re-insert heap.
              heap_dequeue(&loop->timers);
              if (timer->event_type == HEVENT_TYPE_TIMEOUT) {
                  while (timer->next_timeout <= now_hrtime) {
                      timer->next_timeout += (uint64_t)((htimeout_t*)timer)->timeout * 1000;
                  }
              }
              else if (timer->event_type == HEVENT_TYPE_PERIOD) {
                  hperiod_t* period = (hperiod_t*)timer;
                  timer->next_timeout = (uint64_t)cron_next_timeout(period->minute, period->hour, period->day,
                          period->week, period->month) * 1000000;
              }
              heap_insert(&loop->timers, &timer->node);
          }
          EVENT_PENDING(timer);
          ++ntimers;
      }
      return ntimers;
  }
  
  static int hloop_process_ios(hloop_t* loop, int timeout) {
      // That is to call IO multiplexing function such as select, poll, epoll, etc.
      int nevents = iowatcher_poll_events(loop, timeout);
      if (nevents < 0) {
          hlogd("poll_events error=%d", -nevents);
      }
      return nevents < 0 ? 0 : nevents;
  }
  
  static int hloop_process_pendings(hloop_t* loop) {
      if (loop->npendings == 0) return 0;
  
      hevent_t* cur = NULL;
      hevent_t* next = NULL;
      int ncbs = 0;
      // NOTE: invoke event callback from high to low sorted by priority.
      for (int i = HEVENT_PRIORITY_SIZE-1; i >= 0; --i) {
          cur = loop->pendings[i];
          while (cur) {
              next = cur->pending_next;
              if (cur->pending) {
                  if (cur->active && cur->cb) {
                      cur->cb(cur);
                      ++ncbs;
                  }
                  cur->pending = 0;
                  // NOTE: Now we can safely delete event marked as destroy.
                  if (cur->destroy) {
                      EVENT_DEL(cur);
                  }
              }
              cur = next;
          }
          loop->pendings[i] = NULL;
      }
      loop->npendings = 0;
      return ncbs;
  }
  
  // hloop_process_ios -> hloop_process_timers -> hloop_process_idles -> hloop_process_pendings
  static int hloop_process_events(hloop_t* loop) {
      // ios -> timers -> idles
      int nios, ntimers, nidles;
      nios = ntimers = nidles = 0;
  
      // calc blocktime
      int32_t blocktime = HLOOP_MAX_BLOCK_TIME;
      if (loop->timers.root) {
          hloop_update_time(loop);
          uint64_t next_min_timeout = TIMER_ENTRY(loop->timers.root)->next_timeout;
          int64_t blocktime_us = next_min_timeout - hloop_now_hrtime(loop);
          if (blocktime_us <= 0) goto process_timers;
          blocktime = blocktime_us / 1000;
          ++blocktime;
          blocktime = MIN(blocktime, HLOOP_MAX_BLOCK_TIME);
      }
  
      if (loop->nios) {
          nios = hloop_process_ios(loop, blocktime);
      } else {
          hv_msleep(blocktime);
      }
      hloop_update_time(loop);
      // wakeup by hloop_stop
      if (loop->status == HLOOP_STATUS_STOP) {
          return 0;
      }
  
  process_timers:
      if (loop->ntimers) {
          ntimers = hloop_process_timers(loop);
      }
  
      int npendings = loop->npendings;
      if (npendings == 0) {
          if (loop->nidles) {
              nidles= hloop_process_idles(loop);
          }
      }
      int ncbs = hloop_process_pendings(loop);
      // printd("blocktime=%d nios=%d/%u ntimers=%d/%u nidles=%d/%u nactives=%d npendings=%d ncbs=%d\n",
      //         blocktime, nios, loop->nios, ntimers, loop->ntimers, nidles, loop->nidles,
      //         loop->nactives, npendings, ncbs);
      return ncbs;
  }
  
  static void hloop_stat_timer_cb(htimer_t* timer) {
      hloop_t* loop = timer->loop;
      // hlog_set_level(LOG_LEVEL_DEBUG);
      hlogd("[loop] pid=%ld tid=%ld uptime=%lluus cnt=%llu nactives=%u nios=%d ntimers=%d nidles=%u",
          loop->pid, loop->tid, loop->cur_hrtime - loop->start_hrtime, loop->loop_cnt,
          loop->nactives, loop->nios, loop->ntimers, loop->nidles);
  }
  
  static void sockpair_read_cb(hio_t* io, void* buf, int readbytes) {
      hloop_t* loop = io->loop;
      hevent_t* pev = NULL;
      hevent_t ev;
      for (int i = 0; i < readbytes; ++i) {
          hmutex_lock(&loop->custom_events_mutex);
          if (event_queue_empty(&loop->custom_events)) {
              goto unlock;
          }
          pev = event_queue_front(&loop->custom_events);
          if (pev == NULL) {
              goto unlock;
          }
          ev = *pev;
          event_queue_pop_front(&loop->custom_events);
          // NOTE: unlock before cb, avoid deadlock if hloop_post_event called in cb.
          hmutex_unlock(&loop->custom_events_mutex);
          if (ev.cb) {
              ev.cb(&ev);
          }
      }
      return;
  unlock:
      hmutex_unlock(&loop->custom_events_mutex);
  }
  
  void hloop_post_event(hloop_t* loop, hevent_t* ev) {
      char buf = '1';
  
      if (loop->sockpair[0] == -1 || loop->sockpair[1] == -1) {
          hlogw("socketpair not created!");
          return;
      }
  
      if (ev->loop == NULL) {
          ev->loop = loop;
      }
      if (ev->event_type == 0) {
          ev->event_type = HEVENT_TYPE_CUSTOM;
      }
      if (ev->event_id == 0) {
          ev->event_id = hloop_next_event_id();
      }
  
      hmutex_lock(&loop->custom_events_mutex);
      hwrite(loop, loop->sockpair[SOCKPAIR_WRITE_INDEX], &buf, 1, NULL);
      event_queue_push_back(&loop->custom_events, ev);
      hmutex_unlock(&loop->custom_events_mutex);
  }
  
  static void hloop_init(hloop_t* loop) {
  #ifdef OS_WIN
      static int s_wsa_initialized = 0;
      if (s_wsa_initialized == 0) {
          s_wsa_initialized = 1;
          WSADATA wsadata;
          WSAStartup(MAKEWORD(2,2), &wsadata);
      }
  #endif
  #ifdef SIGPIPE
      // NOTE: if not ignore SIGPIPE, write twice when peer close will lead to exit process by SIGPIPE.
      signal(SIGPIPE, SIG_IGN);
  #endif
  
      loop->status = HLOOP_STATUS_STOP;
      loop->pid = hv_getpid();
      loop->tid = hv_gettid();
  
      // idles
      list_init(&loop->idles);
  
      // timers
      heap_init(&loop->timers, timers_compare);
  
      // ios
      io_array_init(&loop->ios, IO_ARRAY_INIT_SIZE);
  
      // readbuf
      loop->readbuf.len = HLOOP_READ_BUFSIZE;
      HV_ALLOC(loop->readbuf.base, loop->readbuf.len);
  
      // iowatcher
      iowatcher_init(loop);
  
      // custom_events
      hmutex_init(&loop->custom_events_mutex);
      event_queue_init(&loop->custom_events, CUSTOM_EVENT_QUEUE_INIT_SIZE);
      loop->sockpair[0] = loop->sockpair[1] = -1;
      if (Socketpair(AF_INET, SOCK_STREAM, 0, loop->sockpair) != 0) {
          hloge("socketpair create failed!");
      }
  
      // NOTE: init start_time here, because htimer_add use it.
      loop->start_ms = gettimeofday_ms();
      loop->start_hrtime = loop->cur_hrtime = gethrtime_us();
  }
  
  static void hloop_cleanup(hloop_t* loop) {
      // pendings
      printd("cleanup pendings...\n");
      for (int i = 0; i < HEVENT_PRIORITY_SIZE; ++i) {
          loop->pendings[i] = NULL;
      }
  
      // ios
      printd("cleanup ios...\n");
      for (int i = 0; i < loop->ios.maxsize; ++i) {
          hio_t* io = loop->ios.ptr[i];
          if (io) {
              hio_free(io);
          }
      }
      io_array_cleanup(&loop->ios);
  
      // idles
      printd("cleanup idles...\n");
      struct list_node* node = loop->idles.next;
      hidle_t* idle;
      while (node != &loop->idles) {
          idle = IDLE_ENTRY(node);
          node = node->next;
          HV_FREE(idle);
      }
      list_init(&loop->idles);
  
      // timers
      printd("cleanup timers...\n");
      htimer_t* timer;
      while (loop->timers.root) {
          timer = TIMER_ENTRY(loop->timers.root);
          heap_dequeue(&loop->timers);
          HV_FREE(timer);
      }
      heap_init(&loop->timers, NULL);
  
      // readbuf
      if (loop->readbuf.base && loop->readbuf.len) {
          HV_FREE(loop->readbuf.base);
          loop->readbuf.base = NULL;
          loop->readbuf.len = 0;
      }
  
      // iowatcher
      iowatcher_cleanup(loop);
  
      // custom_events
      hmutex_lock(&loop->custom_events_mutex);
      if (loop->sockpair[0] != -1 && loop->sockpair[1] != -1) {
          closesocket(loop->sockpair[0]);
          closesocket(loop->sockpair[1]);
          loop->sockpair[0] = loop->sockpair[1] = -1;
      }
      event_queue_cleanup(&loop->custom_events);
      hmutex_unlock(&loop->custom_events_mutex);
      hmutex_destroy(&loop->custom_events_mutex);
  }
  
  hloop_t* hloop_new(int flags) {
      hloop_t* loop;
      HV_ALLOC_SIZEOF(loop);
      hloop_init(loop);
      loop->flags |= flags;
      return loop;
  }
  
  void hloop_free(hloop_t** pp) {
      if (pp && *pp) {
          hloop_cleanup(*pp);
          HV_FREE(*pp);
          *pp = NULL;
      }
  }
  
  // while(loop->status) { hloop_process_events(loop); }
  int hloop_run(hloop_t* loop) {
      if (loop == NULL) return -1;
      if (loop->status == HLOOP_STATUS_RUNNING) return -2;
      loop->status = HLOOP_STATUS_RUNNING;
      loop->pid = hv_getpid();
      loop->tid = hv_gettid();
  
      if (loop->intern_nevents == 0) {
          if (loop->sockpair[0] != -1 && loop->sockpair[1] != -1) {
              hread(loop, loop->sockpair[SOCKPAIR_READ_INDEX], loop->readbuf.base, loop->readbuf.len, sockpair_read_cb);
              ++loop->intern_nevents;
          }
  #ifdef DEBUG
          htimer_add(loop, hloop_stat_timer_cb, HLOOP_STAT_TIMEOUT, INFINITE);
          ++loop->intern_nevents;
  #endif
      }
  
      while (loop->status != HLOOP_STATUS_STOP) {
          if (loop->status == HLOOP_STATUS_PAUSE) {
              hv_msleep(HLOOP_PAUSE_TIME);
              hloop_update_time(loop);
              continue;
          }
          ++loop->loop_cnt;
          if (loop->nactives <= loop->intern_nevents && loop->flags & HLOOP_FLAG_QUIT_WHEN_NO_ACTIVE_EVENTS) {
              break;
          }
          hloop_process_events(loop);
          if (loop->flags & HLOOP_FLAG_RUN_ONCE) {
              break;
          }
      }
      loop->status = HLOOP_STATUS_STOP;
      loop->end_hrtime = gethrtime_us();
  
      if (loop->flags & HLOOP_FLAG_AUTO_FREE) {
          hloop_cleanup(loop);
          HV_FREE(loop);
      }
      return 0;
  }
  
  int hloop_wakeup(hloop_t* loop) {
      hevent_t ev;
      memset(&ev, 0, sizeof(ev));
      hloop_post_event(loop, &ev);
      return 0;
  }
  
  static void hloop_stop_event_cb(hevent_t* ev) {
      ev->loop->status = HLOOP_STATUS_STOP;
  }
  
  int hloop_stop(hloop_t* loop) {
      loop->status = HLOOP_STATUS_STOP;
      if (hv_gettid() != loop->tid) {
          hevent_t ev;
          memset(&ev, 0, sizeof(ev));
          ev.priority = HEVENT_HIGHEST_PRIORITY;
          ev.cb = hloop_stop_event_cb;
          hloop_post_event(loop, &ev);
      }
      return 0;
  }
  
  int hloop_pause(hloop_t* loop) {
      if (loop->status == HLOOP_STATUS_RUNNING) {
          loop->status = HLOOP_STATUS_PAUSE;
      }
      return 0;
  }
  
  int hloop_resume(hloop_t* loop) {
      if (loop->status == HLOOP_STATUS_PAUSE) {
          loop->status = HLOOP_STATUS_RUNNING;
      }
      return 0;
  }
  
  hloop_status_e hloop_status(hloop_t* loop) {
      return loop->status;
  }
  
  void hloop_update_time(hloop_t* loop) {
      loop->cur_hrtime = gethrtime_us();
      if (ABS((int64_t)hloop_now(loop) - (int64_t)time(NULL)) > 1) {
          // systemtime changed, we adjust start_ms
          loop->start_ms = gettimeofday_ms() - (loop->cur_hrtime - loop->start_hrtime) / 1000;
      }
  }
  
  uint64_t hloop_now(hloop_t* loop) {
      return loop->start_ms / 1000 + (loop->cur_hrtime - loop->start_hrtime) / 1000000;
  }
  
  uint64_t hloop_now_ms(hloop_t* loop) {
      return loop->start_ms + (loop->cur_hrtime - loop->start_hrtime) / 1000;
  }
  
  uint64_t hloop_now_hrtime(hloop_t* loop) {
      return loop->start_ms * 1000 + (loop->cur_hrtime - loop->start_hrtime);
  }
  
  long hloop_pid(hloop_t* loop) {
      return loop->pid;
  }
  
  long hloop_tid(hloop_t* loop) {
      return loop->tid;
  }
  
  void  hloop_set_userdata(hloop_t* loop, void* userdata) {
      loop->userdata = userdata;
  }
  
  void* hloop_userdata(hloop_t* loop) {
      return loop->userdata;
  }
  
  hidle_t* hidle_add(hloop_t* loop, hidle_cb cb, uint32_t repeat) {
      hidle_t* idle;
      HV_ALLOC_SIZEOF(idle);
      idle->event_type = HEVENT_TYPE_IDLE;
      idle->priority = HEVENT_LOWEST_PRIORITY;
      idle->repeat = repeat;
      list_add(&idle->node, &loop->idles);
      EVENT_ADD(loop, idle, cb);
      loop->nidles++;
      return idle;
  }
  
  static void __hidle_del(hidle_t* idle) {
      if (idle->destroy) return;
      idle->destroy = 1;
      list_del(&idle->node);
      idle->loop->nidles--;
  }
  
  void hidle_del(hidle_t* idle) {
      if (!idle->active) return;
      __hidle_del(idle);
      EVENT_DEL(idle);
  }
  
  htimer_t* htimer_add(hloop_t* loop, htimer_cb cb, uint32_t timeout, uint32_t repeat) {
      if (timeout == 0)   return NULL;
      htimeout_t* timer;
      HV_ALLOC_SIZEOF(timer);
      timer->event_type = HEVENT_TYPE_TIMEOUT;
      timer->priority = HEVENT_HIGHEST_PRIORITY;
      timer->repeat = repeat;
      timer->timeout = timeout;
      hloop_update_time(loop);
      timer->next_timeout = hloop_now_hrtime(loop) + (uint64_t)timeout*1000;
      // NOTE: Limit granularity to 100ms
      if (timeout >= 1000 && timeout % 100 == 0) {
          timer->next_timeout = timer->next_timeout / 100000 * 100000;
      }
      heap_insert(&loop->timers, &timer->node);
      EVENT_ADD(loop, timer, cb);
      loop->ntimers++;
      return (htimer_t*)timer;
  }
  
  void htimer_reset(htimer_t* timer) {
      if (timer->event_type != HEVENT_TYPE_TIMEOUT) {
          return;
      }
      hloop_t* loop = timer->loop;
      htimeout_t* timeout = (htimeout_t*)timer;
      if (timer->destroy) {
          loop->ntimers++;
      } else {
          heap_remove(&loop->timers, &timer->node);
      }
      if (timer->repeat == 0) {
          timer->repeat = 1;
      }
      timer->next_timeout = hloop_now_hrtime(loop) + (uint64_t)timeout->timeout*1000;
      // NOTE: Limit granularity to 100ms
      if (timeout->timeout >= 1000 && timeout->timeout % 100 == 0) {
          timer->next_timeout = timer->next_timeout / 100000 * 100000;
      }
      heap_insert(&loop->timers, &timer->node);
      EVENT_RESET(timer);
  }
  
  htimer_t* htimer_add_period(hloop_t* loop, htimer_cb cb,
                  int8_t minute,  int8_t hour, int8_t day,
                  int8_t week, int8_t month, uint32_t repeat) {
      if (minute > 59 || hour > 23 || day > 31 || week > 6 || month > 12) {
          return NULL;
      }
      hperiod_t* timer;
      HV_ALLOC_SIZEOF(timer);
      timer->event_type = HEVENT_TYPE_PERIOD;
      timer->priority = HEVENT_HIGH_PRIORITY;
      timer->repeat = repeat;
      timer->minute = minute;
      timer->hour   = hour;
      timer->day    = day;
      timer->month  = month;
      timer->week   = week;
      timer->next_timeout = (uint64_t)cron_next_timeout(minute, hour, day, week, month) * 1000000;
      heap_insert(&loop->timers, &timer->node);
      EVENT_ADD(loop, timer, cb);
      loop->ntimers++;
      return (htimer_t*)timer;
  }
  
  static void __htimer_del(htimer_t* timer) {
      if (timer->destroy) return;
      heap_remove(&timer->loop->timers, &timer->node);
      timer->loop->ntimers--;
      timer->destroy = 1;
  }
  
  void htimer_del(htimer_t* timer) {
      if (!timer->active) return;
      __htimer_del(timer);
      EVENT_DEL(timer);
  }
  
  const char* hio_engine() {
  #ifdef EVENT_SELECT
      return  "select";
  #elif defined(EVENT_POLL)
      return  "poll";
  #elif defined(EVENT_EPOLL)
      return  "epoll";
  #elif defined(EVENT_KQUEUE)
      return  "kqueue";
  #elif defined(EVENT_IOCP)
      return  "iocp";
  #elif defined(EVENT_PORT)
      return  "evport";
  #else
      return  "noevent";
  #endif
  }
  
  static void fill_io_type(hio_t* io) {
      int type = 0;
      socklen_t optlen = sizeof(int);
      int ret = getsockopt(io->fd, SOL_SOCKET, SO_TYPE, (char*)&type, &optlen);
      printd("getsockopt SO_TYPE fd=%d ret=%d type=%d errno=%d\n", io->fd, ret, type, socket_errno());
      if (ret == 0) {
          switch (type) {
          case SOCK_STREAM:   io->io_type = HIO_TYPE_TCP; break;
          case SOCK_DGRAM:    io->io_type = HIO_TYPE_UDP; break;
          case SOCK_RAW:      io->io_type = HIO_TYPE_IP;  break;
          default: io->io_type = HIO_TYPE_SOCKET;         break;
          }
      }
      else if (socket_errno() == ENOTSOCK) {
          switch (io->fd) {
          case 0: io->io_type = HIO_TYPE_STDIN;   break;
          case 1: io->io_type = HIO_TYPE_STDOUT;  break;
          case 2: io->io_type = HIO_TYPE_STDERR;  break;
          default: io->io_type = HIO_TYPE_FILE;   break;
          }
      }
      else {
          io->io_type = HIO_TYPE_TCP;
      }
  }
  
  static void hio_socket_init(hio_t* io) {
      // nonblocking
      nonblocking(io->fd);
      // fill io->localaddr io->peeraddr
      if (io->localaddr == NULL) {
          HV_ALLOC(io->localaddr, sizeof(sockaddr_u));
      }
      if (io->peeraddr == NULL) {
          HV_ALLOC(io->peeraddr, sizeof(sockaddr_u));
      }
      socklen_t addrlen = sizeof(sockaddr_u);
      int ret = getsockname(io->fd, io->localaddr, &addrlen);
      printd("getsockname fd=%d ret=%d errno=%d\n", io->fd, ret, socket_errno());
      // NOTE:
      // tcp_server peeraddr set by accept
      // udp_server peeraddr set by recvfrom
      // tcp_client/udp_client peeraddr set by hio_setpeeraddr
      if (io->io_type == HIO_TYPE_TCP || io->io_type == HIO_TYPE_SSL) {
          // tcp acceptfd
          addrlen = sizeof(sockaddr_u);
          ret = getpeername(io->fd, io->peeraddr, &addrlen);
          printd("getpeername fd=%d ret=%d errno=%d\n", io->fd, ret, socket_errno());
      }
  }
  
  void hio_init(hio_t* io) {
      // alloc localaddr,peeraddr when hio_socket_init
      /*
      if (io->localaddr == NULL) {
          HV_ALLOC(io->localaddr, sizeof(sockaddr_u));
      }
      if (io->peeraddr == NULL) {
          HV_ALLOC(io->peeraddr, sizeof(sockaddr_u));
      }
      */
  
      // write_queue init when hwrite try_write failed
      // write_queue_init(&io->write_queue, 4);
  
      hrecursive_mutex_init(&io->write_mutex);
  }
  
  void hio_ready(hio_t* io) {
      if (io->ready) return;
      // flags
      io->ready = 1;
      io->closed = 0;
      io->accept = io->connect = io->connectex = 0;
      io->recv = io->send = 0;
      io->recvfrom = io->sendto = 0;
      io->close = 0;
      // public:
      io->id = hio_next_id();
      io->io_type = HIO_TYPE_UNKNOWN;
      io->error = 0;
      io->events = io->revents = 0;
      // readbuf
      io->alloced_readbuf = 0;
      io->readbuf.base = io->loop->readbuf.base;
      io->readbuf.len = io->loop->readbuf.len;
      io->readbuf.offset = 0;
      // callbacks
      io->read_cb = NULL;
      io->write_cb = NULL;
      io->close_cb = NULL;
      io->accept_cb = NULL;
      io->connect_cb = NULL;
      // timers
      io->connect_timeout = 0;
      io->connect_timer = NULL;
      io->close_timeout = 0;
      io->close_timer = NULL;
      io->keepalive_timeout = 0;
      io->keepalive_timer = NULL;
      io->heartbeat_interval = 0;
      io->heartbeat_fn = NULL;
      io->heartbeat_timer = NULL;
      // upstream
      io->upstream_io = NULL;
      // unpack
      io->unpack_setting = NULL;
      // private:
      io->event_index[0] = io->event_index[1] = -1;
      io->hovlp = NULL;
      io->ssl = NULL;
  
      // io_type
      fill_io_type(io);
      if (io->io_type & HIO_TYPE_SOCKET) {
          hio_socket_init(io);
      }
  }
  
  void hio_done(hio_t* io) {
      if (!io->ready) return;
      io->ready = 0;
  
      hio_del(io, HV_RDWR);
  
      // readbuf
      hio_free_readbuf(io);
  
      // write_queue
      offset_buf_t* pbuf = NULL;
      hrecursive_mutex_lock(&io->write_mutex);
      while (!write_queue_empty(&io->write_queue)) {
          pbuf = write_queue_front(&io->write_queue);
          HV_FREE(pbuf->base);
          write_queue_pop_front(&io->write_queue);
      }
      write_queue_cleanup(&io->write_queue);
      hrecursive_mutex_unlock(&io->write_mutex);
  }
  
  void hio_free(hio_t* io) {
      if (io == NULL) return;
      // NOTE: call hio_done to cleanup write_queue
      hio_done(io);
      // NOTE: call hio_close to call hclose_cb
      hio_close(io);
      hrecursive_mutex_destroy(&io->write_mutex);
      HV_FREE(io->localaddr);
      HV_FREE(io->peeraddr);
      HV_FREE(io);
  }
  
  bool hio_is_opened(hio_t* io) {
      if (io == NULL) return false;
      return io->ready == 1 && io->closed == 0;
  }
  
  bool hio_is_closed(hio_t* io) {
      if (io == NULL) return true;
      return io->ready == 0 && io->closed == 1;
  }
  
  hio_t* hio_get(hloop_t* loop, int fd) {
      if (fd >= loop->ios.maxsize) {
          int newsize = ceil2e(fd);
          io_array_resize(&loop->ios, newsize > fd ? newsize : 2*fd);
      }
  
      hio_t* io = loop->ios.ptr[fd];
      if (io == NULL) {
          HV_ALLOC_SIZEOF(io);
          hio_init(io);
          io->event_type = HEVENT_TYPE_IO;
          io->loop = loop;
          io->fd = fd;
          loop->ios.ptr[fd] = io;
      }
  
      if (!io->ready) {
          hio_ready(io);
      }
  
      return io;
  }
  
  void hio_detach(hio_t* io) {
      hloop_t* loop = io->loop;
      int fd = io->fd;
      assert(loop != NULL && fd < loop->ios.maxsize);
      loop->ios.ptr[fd] = NULL;
  }
  
  void hio_attach(hloop_t* loop, hio_t* io) {
      int fd = io->fd;
      if (fd >= loop->ios.maxsize) {
          int newsize = ceil2e(fd);
          io_array_resize(&loop->ios, newsize > fd ? newsize : 2*fd);
      }
  
      if (loop->ios.ptr[fd] == NULL) {
          io->loop = loop;
          // NOTE: use new_loop readbuf
          io->readbuf.base = loop->readbuf.base;
          io->readbuf.len = loop->readbuf.len;
          loop->ios.ptr[fd] = io;
      }
  }
  
  int hio_add(hio_t* io, hio_cb cb, int events) {
      printd("hio_add fd=%d io->events=%d events=%d\n", io->fd, io->events, events);
  #ifdef OS_WIN
      // Windows iowatcher not work on stdio
      if (io->fd < 3) return -1;
  #endif
      hloop_t* loop = io->loop;
      if (!io->active) {
          EVENT_ADD(loop, io, cb);
          loop->nios++;
      }
  
      if (!io->ready) {
          hio_ready(io);
      }
  
      if (cb) {
          io->cb = (hevent_cb)cb;
      }
  
      if (!(io->events & events)) {
          iowatcher_add_event(loop, io->fd, events);
          io->events |= events;
      }
      return 0;
  }
  
  int hio_del(hio_t* io, int events) {
      printd("hio_del fd=%d io->events=%d events=%d\n", io->fd, io->events, events);
  #ifdef OS_WIN
      // Windows iowatcher not work on stdio
      if (io->fd < 3) return -1;
  #endif
      if (!io->active) return -1;
  
      if (io->events & events) {
          iowatcher_del_event(io->loop, io->fd, events);
          io->events &= ~events;
      }
      if (io->events == 0) {
          io->loop->nios--;
          // NOTE: not EVENT_DEL, avoid free
          EVENT_INACTIVE(io);
      }
      return 0;
  }
  
  static void hio_close_event_cb(hevent_t* ev) {
      hio_t* io = (hio_t*)ev->userdata;
      uint32_t id = (uintptr_t)ev->privdata;
      if (io->id != id) return;
      hio_close(io);
  }
  
  int hio_close_async(hio_t* io) {
      hevent_t ev;
      memset(&ev, 0, sizeof(ev));
      ev.cb = hio_close_event_cb;
      ev.userdata = io;
      ev.privdata = (void*)(uintptr_t)io->id;
      ev.priority = HEVENT_HIGH_PRIORITY;
      hloop_post_event(io->loop, &ev);
      return 0;
  }
  
  hio_t* hread(hloop_t* loop, int fd, void* buf, size_t len, hread_cb read_cb) {
      hio_t* io = hio_get(loop, fd);
      assert(io != NULL);
      if (buf && len) {
          io->readbuf.base = (char*)buf;
          io->readbuf.len = len;
      }
      if (read_cb) {
          io->read_cb = read_cb;
      }
      hio_read(io);
      return io;
  }
  
  hio_t* hwrite(hloop_t* loop, int fd, const void* buf, size_t len, hwrite_cb write_cb) {
      hio_t* io = hio_get(loop, fd);
      assert(io != NULL);
      if (write_cb) {
          io->write_cb = write_cb;
      }
      hio_write(io, buf, len);
      return io;
  }
  
  hio_t* haccept(hloop_t* loop, int listenfd, haccept_cb accept_cb) {
      hio_t* io = hio_get(loop, listenfd);
      assert(io != NULL);
      if (accept_cb) {
          io->accept_cb = accept_cb;
      }
      hio_accept(io);
      return io;
  }
  
  hio_t* hconnect (hloop_t* loop, int connfd, hconnect_cb connect_cb) {
      hio_t* io = hio_get(loop, connfd);
      assert(io != NULL);
      if (connect_cb) {
          io->connect_cb = connect_cb;
      }
      hio_connect(io);
      return io;
  }
  
  void hclose (hloop_t* loop, int fd) {
      hio_t* io = hio_get(loop, fd);
      assert(io != NULL);
      hio_close(io);
  }
  
  hio_t* hrecv (hloop_t* loop, int connfd, void* buf, size_t len, hread_cb read_cb) {
      //hio_t* io = hio_get(loop, connfd);
      //assert(io != NULL);
      //io->recv = 1;
      //if (io->io_type != HIO_TYPE_SSL) {
          //io->io_type = HIO_TYPE_TCP;
      //}
      return hread(loop, connfd, buf, len, read_cb);
  }
  
  hio_t* hsend (hloop_t* loop, int connfd, const void* buf, size_t len, hwrite_cb write_cb) {
      //hio_t* io = hio_get(loop, connfd);
      //assert(io != NULL);
      //io->send = 1;
      //if (io->io_type != HIO_TYPE_SSL) {
          //io->io_type = HIO_TYPE_TCP;
      //}
      return hwrite(loop, connfd, buf, len, write_cb);
  }
  
  hio_t* hrecvfrom (hloop_t* loop, int sockfd, void* buf, size_t len, hread_cb read_cb) {
      //hio_t* io = hio_get(loop, sockfd);
      //assert(io != NULL);
      //io->recvfrom = 1;
      //io->io_type = HIO_TYPE_UDP;
      return hread(loop, sockfd, buf, len, read_cb);
  }
  
  hio_t* hsendto (hloop_t* loop, int sockfd, const void* buf, size_t len, hwrite_cb write_cb) {
      //hio_t* io = hio_get(loop, sockfd);
      //assert(io != NULL);
      //io->sendto = 1;
      //io->io_type = HIO_TYPE_UDP;
      return hwrite(loop, sockfd, buf, len, write_cb);
  }
  
  hio_t* hio_create(hloop_t* loop, const char* host, int port, int type) {
      sockaddr_u peeraddr;
      memset(&peeraddr, 0, sizeof(peeraddr));
      int ret = sockaddr_set_ipport(&peeraddr, host, port);
      if (ret != 0) {
          //printf("unknown host: %s\n", host);
          return NULL;
      }
      int connfd = socket(peeraddr.sa.sa_family, type, 0);
      if (connfd < 0) {
          perror("socket");
          return NULL;
      }
  
      hio_t* io = hio_get(loop, connfd);
      assert(io != NULL);
      hio_set_peeraddr(io, &peeraddr.sa, sockaddr_len(&peeraddr));
      return io;
  }
  
  hio_t* hloop_create_tcp_server (hloop_t* loop, const char* host, int port, haccept_cb accept_cb) {
      int listenfd = Listen(port, host);
      if (listenfd < 0) {
          return NULL;
      }
      hio_t* io = haccept(loop, listenfd, accept_cb);
      if (io == NULL) {
          closesocket(listenfd);
      }
      return io;
  }
  
  hio_t* hloop_create_tcp_client (hloop_t* loop, const char* host, int port, hconnect_cb connect_cb) {
      hio_t* io = hio_create(loop, host, port, SOCK_STREAM);
      if (io == NULL) return NULL;
      hconnect(loop, io->fd, connect_cb);
      return io;
  }
  
  hio_t* hloop_create_ssl_server (hloop_t* loop, const char* host, int port, haccept_cb accept_cb) {
      hio_t* io = hloop_create_tcp_server(loop, host, port, accept_cb);
      if (io == NULL) return NULL;
      hio_enable_ssl(io);
      return io;
  }
  
  hio_t* hloop_create_ssl_client (hloop_t* loop, const char* host, int port, hconnect_cb connect_cb) {
      hio_t* io = hio_create(loop, host, port, SOCK_STREAM);
      if (io == NULL) return NULL;
      hio_enable_ssl(io);
      hconnect(loop, io->fd, connect_cb);
      return io;
  }
  
  hio_t* hloop_create_udp_server(hloop_t* loop, const char* host, int port) {
      int bindfd = Bind(port, host, SOCK_DGRAM);
      if (bindfd < 0) {
          return NULL;
      }
      return hio_get(loop, bindfd);
  }
  
  hio_t* hloop_create_udp_client(hloop_t* loop, const char* host, int port) {
      return hio_create(loop, host, port, SOCK_DGRAM);
  }
  
  // upstream
  void hio_read_upstream(hio_t* io) {
      hio_t* upstream_io = io->upstream_io;
      if (upstream_io) {
          hio_read(io);
          hio_read(upstream_io);
      }
  }
  
  void hio_write_upstream(hio_t* io, void* buf, int bytes) {
      hio_t* upstream_io = io->upstream_io;
      if (upstream_io) {
          hio_write(upstream_io, buf, bytes);
      }
  }
  
  void hio_close_upstream(hio_t* io) {
      hio_t* upstream_io = io->upstream_io;
      if (upstream_io) {
          hio_close(upstream_io);
      }
  }
  
  void hio_setup_upstream(hio_t* io1, hio_t* io2) {
      io1->upstream_io = io2;
      io2->upstream_io = io1;
      hio_setcb_read(io1, hio_write_upstream);
      hio_setcb_read(io2, hio_write_upstream);
  }
  
  hio_t* hio_get_upstream(hio_t* io) {
      return io->upstream_io;
  }
  
  hio_t* hio_setup_tcp_upstream(hio_t* io, const char* host, int port, int ssl) {
      hio_t* upstream_io = hio_create(io->loop, host, port, SOCK_STREAM);
      if (upstream_io == NULL) return NULL;
      if (ssl) hio_enable_ssl(upstream_io);
      hio_setup_upstream(io, upstream_io);
      hio_setcb_close(io, hio_close_upstream);
      hio_setcb_close(upstream_io, hio_close_upstream);
      hconnect(io->loop, upstream_io->fd, hio_read_upstream);
      return upstream_io;
  }
  
  hio_t* hio_setup_udp_upstream(hio_t* io, const char* host, int port) {
      hio_t* upstream_io = hio_create(io->loop, host, port, SOCK_DGRAM);
      if (upstream_io == NULL) return NULL;
      hio_setup_upstream(io, upstream_io);
      hio_read_upstream(io);
      return upstream_io;
  }
  