oceanbase

1
/**
2
 * Copyright (c) 2021 OceanBase
3
 * OceanBase CE is licensed under Mulan PubL v2.
4
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
5
 * You may obtain a copy of Mulan PubL v2 at:
6
 *          http://license.coscl.org.cn/MulanPubL-2.0
7
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
8
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
9
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
10
 * See the Mulan PubL v2 for more details.
11
 */
12

13
int64_t ob_update_loop_ts();
14
struct epoll_event *ussl_make_epoll_event(struct epoll_event *event, uint32_t event_flag, void *val)
15
{
16
  event->events = event_flag;
17
  event->data.ptr = val;
18
  return event;
19
}
20

21
int ussl_eloop_init(ussl_eloop_t *ep)
22
{
23
  ep->fd = epoll_create1(EPOLL_CLOEXEC);
24
  ussl_dlink_init(&ep->ready_link);
25
  return (ep->fd < 0) ? errno : 0;
26
}
27

28
int ussl_eloop_regist(ussl_eloop_t *ep, ussl_sock_t *s, uint32_t eflag)
29
{
30
  int err = 0;
31
  struct epoll_event event;
32
  uint32_t flag = eflag | EPOLLERR;
33
  s->mask = 0;
34
  s->ready_link.next = NULL;
35
  if (0 != libc_epoll_ctl(ep->fd, EPOLL_CTL_ADD, s->fd, ussl_make_epoll_event(&event, flag, s))) {
36
    err = -EIO;
37
    ussl_log_error("epoll_ctl add failed, epfd:%d, fd:%d, errno:%d", ep->fd, s->fd, errno);
38
  } else {
39
    ussl_log_info("sock regist: %p fd=%d", s, s->fd);
40
  }
41
  return err;
42
}
43

44
static void ussl_eloop_fire(ussl_eloop_t *ep, ussl_sock_t *s)
45
{
46
  if (!s->ready_link.next) {
47
    ussl_dlink_insert(&ep->ready_link, &s->ready_link);
48
  } else {
49
    ussl_sks(s, PENDING);
50
  }
51
}
52

53
static void ussl_eloop_refire(ussl_eloop_t *ep, int64_t epoll_timeout)
54
{
55
  const int maxevents = 512;
56
  struct epoll_event events[maxevents];
57
  int cnt = ob_epoll_wait(ep->fd, events, maxevents, epoll_timeout);
58
  for (int i = 0; i < cnt; i++) {
59
    ussl_sock_t *s = (ussl_sock_t *)events[i].data.ptr;
60
    s->mask |= events[i].events;
61
    ussl_eloop_fire(ep, s);
62
  }
63
}
64

65
static void ussl_sock_destroy(ussl_sock_t *s)
66
{
67
  ussl_dlink_delete(&s->ready_link);
68
  if (s->fty) {
69
    s->fty->destroy(s->fty, s);
70
  }
71
}
72

73
static void ussl_eloop_handle_sock_event(ussl_sock_t *s)
74
{
75
  int err = 0;
76
  if (ussl_skt(s, ERR) || ussl_skt(s, HUP)) {
77
    ussl_log_info("sock has error: sock:%p, fd:%d, mask:0x%x", s, s->fd, s->mask);
78
    s->has_error = 1;
79
    ussl_sock_destroy(s);
80
  } else if (0 == (err = s->handle_event(s))) {
81
    // yield
82
  } else if (EAGAIN == err) {
83
    if (ussl_skt(s, PENDING)) {
84
      ussl_skc(s, PENDING);
85
    } else {
86
      ussl_dlink_delete(&s->ready_link);
87
    }
88
  } else {
89
    ussl_sock_destroy(s);
90
  }
91
}
92

93
int ussl_eloop_run(ussl_eloop_t *ep)
94
{
95
  while (!ussl_is_stop()) {
96
    ob_update_loop_ts();
97
    int64_t epoll_timeout = 1000;
98
    if (ep->ready_link.next != &(ep->ready_link)) {
99
      epoll_timeout = 0;
100
    }
101
    ussl_eloop_refire(ep, epoll_timeout);
102
    ussl_dlink_for(&ep->ready_link, p) { ussl_eloop_handle_sock_event(ussl_structof(p, ussl_sock_t, ready_link)); }
103
    check_and_handle_timeout_event();
104
  }
105
  close(ep->fd);
106
  return 0;
107
}
108