fd.cpp 17.7 KB
Newer Older
gejun's avatar
gejun committed
1
// bthread - A M:N threading library to make applications more concurrent.
gejun's avatar
gejun committed
2
// Copyright (c) 2014 Baidu, Inc.
gejun's avatar
gejun committed
3 4 5 6 7 8 9 10 11 12 13 14
// 
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// 
//     http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
gejun's avatar
gejun committed
15 16 17 18

// Author: Ge,Jun (gejun@baidu.com)
// Date: Thu Aug  7 18:56:27 CST 2014

gejun's avatar
gejun committed
19
#include "butil/compat.h"
gejun's avatar
gejun committed
20 21
#include <new>                                   // std::nothrow
#include <sys/poll.h>                            // poll()
zhujiashun's avatar
zhujiashun committed
22
#if defined(OS_MACOSX)
23 24
#include <sys/types.h>                           // struct kevent
#include <sys/event.h>                           // kevent(), kqueue()
zhujiashun's avatar
zhujiashun committed
25
#endif
26 27 28 29 30
#include "butil/atomicops.h"
#include "butil/time.h"
#include "butil/fd_utility.h"                     // make_non_blocking
#include "butil/logging.h"
#include "butil/third_party/murmurhash3/murmurhash3.h"   // fmix32
gejun's avatar
gejun committed
31 32 33 34 35 36 37 38 39 40 41 42 43
#include "bthread/butex.h"                       // butex_*
#include "bthread/task_group.h"                  // TaskGroup
#include "bthread/bthread.h"                             // bthread_start_urgent

// Implement bthread functions on file descriptors

namespace bthread {

extern BAIDU_THREAD_LOCAL TaskGroup* tls_task_group;

template <typename T, size_t NBLOCK, size_t BLOCK_SIZE>
class LazyArray {
    struct Block {
44
        butil::atomic<T> items[BLOCK_SIZE];
gejun's avatar
gejun committed
45 46 47 48
    };

public:
    LazyArray() {
49
        memset(_blocks, 0, sizeof(butil::atomic<Block*>) * NBLOCK);
gejun's avatar
gejun committed
50 51
    }

52
    butil::atomic<T>* get_or_new(size_t index) {
gejun's avatar
gejun committed
53 54 55 56 57
        const size_t block_index = index / BLOCK_SIZE;
        if (block_index >= NBLOCK) {
            return NULL;
        }
        const size_t block_offset = index - block_index * BLOCK_SIZE;
58
        Block* b = _blocks[block_index].load(butil::memory_order_consume);
gejun's avatar
gejun committed
59 60 61 62 63
        if (b != NULL) {
            return b->items + block_offset;
        }
        b = new (std::nothrow) Block;
        if (NULL == b) {
64
            b = _blocks[block_index].load(butil::memory_order_consume);
gejun's avatar
gejun committed
65 66 67 68 69 70
            return (b ? b->items + block_offset : NULL);
        }
        // Set items to default value of T.
        std::fill(b->items, b->items + BLOCK_SIZE, T());
        Block* expected = NULL;
        if (_blocks[block_index].compare_exchange_strong(
71 72
                expected, b, butil::memory_order_release,
                butil::memory_order_consume)) {
gejun's avatar
gejun committed
73 74 75 76 77 78
            return b->items + block_offset;
        }
        delete b;
        return expected->items + block_offset;
    }

79
    butil::atomic<T>* get(size_t index) const {
gejun's avatar
gejun committed
80 81 82
        const size_t block_index = index / BLOCK_SIZE;
        if (__builtin_expect(block_index < NBLOCK, 1)) {
            const size_t block_offset = index - block_index * BLOCK_SIZE;
83
            Block* const b = _blocks[block_index].load(butil::memory_order_consume);
gejun's avatar
gejun committed
84 85 86 87 88 89 90 91
            if (__builtin_expect(b != NULL, 1)) {
                return b->items + block_offset;
            }
        }
        return NULL;
    }

private:
92
    butil::atomic<Block*> _blocks[NBLOCK];
gejun's avatar
gejun committed
93 94
};

95
typedef butil::atomic<int> EpollButex;
gejun's avatar
gejun committed
96 97 98 99

static EpollButex* const CLOSING_GUARD = (EpollButex*)(intptr_t)-1L;

#ifndef NDEBUG
100
butil::static_atomic<int> break_nums = BUTIL_STATIC_ATOMIC_INIT(0);
gejun's avatar
gejun committed
101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119
#endif

// Able to address 67108864 file descriptors, should be enough.
LazyArray<EpollButex*, 262144/*NBLOCK*/, 256/*BLOCK_SIZE*/> fd_butexes;

static const int BTHREAD_DEFAULT_EPOLL_SIZE = 65536;

class EpollThread {
public:
    EpollThread()
        : _epfd(-1)
        , _stop(false)
        , _tid(0) {
    }

    int start(int epoll_size) {
        if (started()) {
            return -1;
        }
120 121 122 123 124 125
        _start_mutex.lock();
        // Double check
        if (started()) {
            _start_mutex.unlock();
            return -1;
        }
126
#if defined(OS_LINUX)
gejun's avatar
gejun committed
127
        _epfd = epoll_create(epoll_size);
128 129 130
#elif defined(OS_MACOSX)
        _epfd = kqueue();
#endif
131
        _start_mutex.unlock();
gejun's avatar
gejun committed
132
        if (_epfd < 0) {
133
            PLOG(FATAL) << "Fail to epoll_create/kqueue";
gejun's avatar
gejun committed
134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169
            return -1;
        }
        if (bthread_start_background(
                &_tid, NULL, EpollThread::run_this, this) != 0) {
            close(_epfd);
            _epfd = -1;
            LOG(FATAL) << "Fail to create epoll bthread";
            return -1;
        }
        return 0;
    }

    // Note: This function does not wake up suspended fd_wait. This is fine
    // since stop_and_join is only called on program's termination
    // (g_task_control.stop()), suspended bthreads do not block quit of
    // worker pthreads and completion of g_task_control.stop().
    int stop_and_join() {
        if (!started()) {
            return 0;
        }
        // No matter what this function returns, _epfd will be set to -1
        // (making started() false) to avoid latter stop_and_join() to
        // enter again.
        const int saved_epfd = _epfd;
        _epfd = -1;

        // epoll_wait cannot be woken up by closing _epfd. We wake up
        // epoll_wait by inserting a fd continuously triggering EPOLLOUT.
        // Visibility of _stop: constant EPOLLOUT forces epoll_wait to see
        // _stop (to be true) finally.
        _stop = true;
        int closing_epoll_pipe[2];
        if (pipe(closing_epoll_pipe)) {
            PLOG(FATAL) << "Fail to create closing_epoll_pipe";
            return -1;
        }
170
#if defined(OS_LINUX)
gejun's avatar
gejun committed
171 172 173
        epoll_event evt = { EPOLLOUT, { NULL } };
        if (epoll_ctl(saved_epfd, EPOLL_CTL_ADD,
                      closing_epoll_pipe[1], &evt) < 0) {
174 175 176 177 178 179
#elif defined(OS_MACOSX)
        struct kevent kqueue_event;
        EV_SET(&kqueue_event, closing_epoll_pipe[1], EVFILT_WRITE, EV_ADD | EV_ENABLE,
                0, 0, NULL);
        if (kevent(saved_epfd, &kqueue_event, 1, NULL, 0, NULL) < 0) {
#endif
gejun's avatar
gejun committed
180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195
            PLOG(FATAL) << "Fail to add closing_epoll_pipe into epfd="
                        << saved_epfd;
            return -1;
        }

        const int rc = bthread_join(_tid, NULL);
        if (rc) {
            LOG(FATAL) << "Fail to join EpollThread, " << berror(rc);
            return -1;
        }
        close(closing_epoll_pipe[0]);
        close(closing_epoll_pipe[1]);
        close(saved_epfd);
        return 0;
    }

196
    int fd_wait(int fd, unsigned events, const timespec* abstime) {
197
        butil::atomic<EpollButex*>* p = fd_butexes.get_or_new(fd);
gejun's avatar
gejun committed
198 199 200 201 202
        if (NULL == p) {
            errno = ENOMEM;
            return -1;
        }

203
        EpollButex* butex = p->load(butil::memory_order_consume);
gejun's avatar
gejun committed
204 205 206 207 208
        if (NULL == butex) {
            // It is rare to wait on one file descriptor from multiple threads
            // simultaneously. Creating singleton by optimistic locking here
            // saves mutexes for each butex.
            butex = butex_create_checked<EpollButex>();
209
            butex->store(0, butil::memory_order_relaxed);
gejun's avatar
gejun committed
210 211
            EpollButex* expected = NULL;
            if (!p->compare_exchange_strong(expected, butex,
212 213
                                            butil::memory_order_release,
                                            butil::memory_order_consume)) {
gejun's avatar
gejun committed
214 215 216 217 218 219 220 221 222
                butex_destroy(butex);
                butex = expected;
            }
        }
        
        while (butex == CLOSING_GUARD) {  // bthread_close() is running.
            if (sched_yield() < 0) {
                return -1;
            }
223
            butex = p->load(butil::memory_order_consume);
gejun's avatar
gejun committed
224 225 226 227
        }
        // Save value of butex before adding to epoll because the butex may
        // be changed before butex_wait. No memory fence because EPOLL_CTL_MOD
        // and EPOLL_CTL_ADD shall have release fence.
228
        const int expected_val = butex->load(butil::memory_order_relaxed);
gejun's avatar
gejun committed
229

230 231 232
#if defined(OS_LINUX)
# ifdef BAIDU_KERNEL_FIXED_EPOLLONESHOT_BUG
        epoll_event evt = { events | EPOLLONESHOT, { butex } };
gejun's avatar
gejun committed
233 234 235 236 237 238 239
        if (epoll_ctl(_epfd, EPOLL_CTL_MOD, fd, &evt) < 0) {
            if (epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &evt) < 0 &&
                    errno != EEXIST) {
                PLOG(FATAL) << "Fail to add fd=" << fd << " into epfd=" << _epfd;
                return -1;
            }
        }
240
# else
gejun's avatar
gejun committed
241
        epoll_event evt;
242
        evt.events = events;
gejun's avatar
gejun committed
243 244 245 246 247 248
        evt.data.fd = fd;
        if (epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &evt) < 0 &&
            errno != EEXIST) {
            PLOG(FATAL) << "Fail to add fd=" << fd << " into epfd=" << _epfd;
            return -1;
        }
249 250 251 252 253 254 255 256 257 258
# endif
#elif defined(OS_MACOSX)
        struct kevent kqueue_event;
        EV_SET(&kqueue_event, fd, events, EV_ADD | EV_ENABLE | EV_ONESHOT,
                0, 0, butex);
        if (kevent(_epfd, &kqueue_event, 1, NULL, 0, NULL) < 0) {
            PLOG(FATAL) << "Fail to add fd=" << fd << " into kqueuefd=" << _epfd;
            return -1;
        }
#endif
259 260 261
        if (butex_wait(butex, expected_val, abstime) < 0 &&
            errno != EWOULDBLOCK && errno != EINTR) {
            return -1;
gejun's avatar
gejun committed
262
        }
263
        return 0;
gejun's avatar
gejun committed
264 265 266 267 268 269 270 271
    }

    int fd_close(int fd) {
        if (fd < 0) {
            // what close(-1) returns
            errno = EBADF;
            return -1;
        }
272
        butil::atomic<EpollButex*>* pbutex = bthread::fd_butexes.get(fd);
gejun's avatar
gejun committed
273 274 275 276 277
        if (NULL == pbutex) {
            // Did not call bthread_fd functions, close directly.
            return close(fd);
        }
        EpollButex* butex = pbutex->exchange(
278
            CLOSING_GUARD, butil::memory_order_relaxed);
gejun's avatar
gejun committed
279 280 281 282 283 284
        if (butex == CLOSING_GUARD) {
            // concurrent double close detected.
            errno = EBADF;
            return -1;
        }
        if (butex != NULL) {
285
            butex->fetch_add(1, butil::memory_order_relaxed);
gejun's avatar
gejun committed
286 287
            butex_wake_all(butex);
        }
288
#if defined(OS_LINUX)
gejun's avatar
gejun committed
289
        epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, NULL);
290 291 292 293 294 295 296
#elif defined(OS_MACOSX)
        struct kevent evt;
        EV_SET(&evt, fd, EVFILT_WRITE, EV_DELETE, 0, 0, NULL);
        kevent(_epfd, &evt, 1, NULL, 0, NULL);
        EV_SET(&evt, fd, EVFILT_READ, EV_DELETE, 0, 0, NULL);
        kevent(_epfd, &evt, 1, NULL, 0, NULL);
#endif
gejun's avatar
gejun committed
297
        const int rc = close(fd);
298
        pbutex->exchange(butex, butil::memory_order_relaxed);
gejun's avatar
gejun committed
299 300 301 302 303 304 305 306 307 308 309 310 311 312 313
        return rc;
    }

    bool started() const {
        return _epfd >= 0;
    }

private:
    static void* run_this(void* arg) {
        return static_cast<EpollThread*>(arg)->run();
    }

    void* run() {
        const int initial_epfd = _epfd;
        const size_t MAX_EVENTS = 32;
314
#if defined(OS_LINUX)
gejun's avatar
gejun committed
315
        epoll_event* e = new (std::nothrow) epoll_event[MAX_EVENTS];
316 317 318 319
#elif defined(OS_MACOSX)
        typedef struct kevent KEVENT;
        struct kevent* e = new (std::nothrow) KEVENT[MAX_EVENTS];
#endif
gejun's avatar
gejun committed
320 321 322 323 324
        if (NULL == e) {
            LOG(FATAL) << "Fail to new epoll_event";
            return NULL;
        }

325 326
#if defined(OS_LINUX)
# ifndef BAIDU_KERNEL_FIXED_EPOLLONESHOT_BUG
gejun's avatar
gejun committed
327
        DLOG(INFO) << "Use DEL+ADD instead of EPOLLONESHOT+MOD due to kernel bug. Performance will be much lower.";
328
# endif
gejun's avatar
gejun committed
329 330 331
#endif
        while (!_stop) {
            const int epfd = _epfd;
332
#if defined(OS_LINUX)
gejun's avatar
gejun committed
333
            const int n = epoll_wait(epfd, e, MAX_EVENTS, -1);
334 335 336
#elif defined(OS_MACOSX)
            const int n = kevent(epfd, NULL, 0, e, MAX_EVENTS, NULL);
#endif
gejun's avatar
gejun committed
337 338 339 340 341 342 343
            if (_stop) {
                break;
            }

            if (n < 0) {
                if (errno == EINTR) {
#ifndef NDEBUG
344
                    break_nums.fetch_add(1, butil::memory_order_relaxed);
gejun's avatar
gejun committed
345 346 347 348 349 350 351 352 353 354 355 356 357
                    int* p = &errno;
                    const char* b = berror();
                    const char* b2 = berror(errno);
                    DLOG(FATAL) << "Fail to epoll epfd=" << epfd << ", "
                                << errno << " " << p << " " <<  b << " " <<  b2;
#endif
                    continue;
                }

                PLOG(INFO) << "Fail to epoll epfd=" << epfd;
                break;
            }

358 359
#if defined(OS_LINUX)
# ifndef BAIDU_KERNEL_FIXED_EPOLLONESHOT_BUG
gejun's avatar
gejun committed
360 361 362
            for (int i = 0; i < n; ++i) {
                epoll_ctl(epfd, EPOLL_CTL_DEL, e[i].data.fd, NULL);
            }
363
# endif
gejun's avatar
gejun committed
364 365
#endif
            for (int i = 0; i < n; ++i) {
366 367
#if defined(OS_LINUX)
# ifdef BAIDU_KERNEL_FIXED_EPOLLONESHOT_BUG
gejun's avatar
gejun committed
368
                EpollButex* butex = static_cast<EpollButex*>(e[i].data.ptr);
369
# else
370
                butil::atomic<EpollButex*>* pbutex = fd_butexes.get(e[i].data.fd);
gejun's avatar
gejun committed
371
                EpollButex* butex = pbutex ?
372
                    pbutex->load(butil::memory_order_consume) : NULL;
373 374 375
# endif
#elif defined(OS_MACOSX)
                EpollButex* butex = static_cast<EpollButex*>(e[i].udata);
gejun's avatar
gejun committed
376 377
#endif
                if (butex != NULL && butex != CLOSING_GUARD) {
378
                    butex->fetch_add(1, butil::memory_order_relaxed);
gejun's avatar
gejun committed
379 380 381 382 383 384 385 386 387 388 389 390 391 392
                    butex_wake_all(butex);
                }
            }
        }

        delete [] e;
        DLOG(INFO) << "EpollThread=" << _tid << "(epfd="
                   << initial_epfd << ") is about to stop";
        return NULL;
    }

    int _epfd;
    bool _stop;
    bthread_t _tid;
393
    butil::Mutex _start_mutex;
gejun's avatar
gejun committed
394 395 396 397 398 399 400 401 402 403 404
};

EpollThread epoll_thread[BTHREAD_EPOLL_THREAD_NUM];

static inline EpollThread& get_epoll_thread(int fd) {
    if (BTHREAD_EPOLL_THREAD_NUM == 1UL) {
        EpollThread& et = epoll_thread[0];
        et.start(BTHREAD_DEFAULT_EPOLL_SIZE);
        return et;
    }

405
    EpollThread& et = epoll_thread[butil::fmix32(fd) % BTHREAD_EPOLL_THREAD_NUM];
gejun's avatar
gejun committed
406 407 408 409
    et.start(BTHREAD_DEFAULT_EPOLL_SIZE);
    return et;
}

410
//TODO(zhujiashun): change name
gejun's avatar
gejun committed
411 412 413 414 415 416 417 418 419 420 421
int stop_and_join_epoll_threads() {
    // Returns -1 if any epoll thread failed to stop.
    int rc = 0;
    for (size_t i = 0; i < BTHREAD_EPOLL_THREAD_NUM; ++i) {
        if (epoll_thread[i].stop_and_join() < 0) {
            rc = -1;
        }
    }
    return rc;
}

422
#if defined(OS_LINUX)
gejun's avatar
gejun committed
423 424 425 426 427 428 429 430 431 432
short epoll_to_poll_events(uint32_t epoll_events) {
    // Most POLL* and EPOLL* are same values.
    short poll_events = (epoll_events &
                         (EPOLLIN | EPOLLPRI | EPOLLOUT |
                          EPOLLRDNORM | EPOLLRDBAND |
                          EPOLLWRNORM | EPOLLWRBAND |
                          EPOLLMSG | EPOLLERR | EPOLLHUP));
    CHECK_EQ((uint32_t)poll_events, epoll_events);
    return poll_events;
}
433 434 435 436 437 438 439 440 441 442 443 444 445
#elif defined(OS_MACOSX)
short kqueue_to_poll_events(uint32_t kqueue_events) {
    //TODO: add more values?
    short poll_events = 0;
    if (kqueue_events == EVFILT_READ) {
        poll_events |= POLLIN;
    }
    if (kqueue_events == EVFILT_WRITE) {
        poll_events |= POLLOUT;
    }
    return poll_events;
}
#endif
gejun's avatar
gejun committed
446 447

// For pthreads.
448
int pthread_fd_wait(int fd, unsigned events,
gejun's avatar
gejun committed
449 450 451 452 453
                    const timespec* abstime) {
    int diff_ms = -1;
    if (abstime) {
        timespec now;
        clock_gettime(CLOCK_REALTIME, &now);
454 455
        int64_t now_us = butil::timespec_to_microseconds(now);
        int64_t abstime_us = butil::timespec_to_microseconds(*abstime);
gejun's avatar
gejun committed
456 457 458 459 460 461
        if (abstime_us <= now_us) {
            errno = ETIMEDOUT;
            return -1;
        }
        diff_ms = (abstime_us - now_us + 999L) / 1000L;
    }
462
#if defined(OS_LINUX)
463
    const short poll_events = bthread::epoll_to_poll_events(events);
464
#elif defined(OS_MACOSX)
465
    const short poll_events = bthread::kqueue_to_poll_events(events);
466
#endif
gejun's avatar
gejun committed
467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490
    if (poll_events == 0) {
        errno = EINVAL;
        return -1;
    }
    pollfd ufds = { fd, poll_events, 0 };
    const int rc = poll(&ufds, 1, diff_ms);
    if (rc < 0) {
        return -1;
    }
    if (rc == 0) {
        errno = ETIMEDOUT;
        return -1;
    }
    if (ufds.revents & POLLNVAL) {
        errno = EBADF;
        return -1;
    }
    return 0;
}

}  // namespace bthread

extern "C" {

491
int bthread_fd_wait(int fd, unsigned events) {
gejun's avatar
gejun committed
492 493 494 495 496 497 498
    if (fd < 0) {
        errno = EINVAL;
        return -1;
    }
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (NULL != g && !g->is_current_pthread_task()) {
        return bthread::get_epoll_thread(fd).fd_wait(
499
            fd, events, NULL);
gejun's avatar
gejun committed
500
    }
501
    return bthread::pthread_fd_wait(fd, events, NULL);
gejun's avatar
gejun committed
502 503
}

504
int bthread_fd_timedwait(int fd, unsigned events,
gejun's avatar
gejun committed
505
                         const timespec* abstime) {
gejun's avatar
gejun committed
506
    if (NULL == abstime) {
507
        return bthread_fd_wait(fd, events);
gejun's avatar
gejun committed
508 509 510 511 512 513 514 515
    }
    if (fd < 0) {
        errno = EINVAL;
        return -1;
    }
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (NULL != g && !g->is_current_pthread_task()) {
        return bthread::get_epoll_thread(fd).fd_wait(
516
            fd, events, abstime);
gejun's avatar
gejun committed
517
    }
518
    return bthread::pthread_fd_wait(fd, events, abstime);
gejun's avatar
gejun committed
519 520 521
}

int bthread_connect(int sockfd, const sockaddr* serv_addr,
gejun's avatar
gejun committed
522
                    socklen_t addrlen) {
gejun's avatar
gejun committed
523 524 525 526 527
    bthread::TaskGroup* g = bthread::tls_task_group;
    if (NULL == g || g->is_current_pthread_task()) {
        return ::connect(sockfd, serv_addr, addrlen);
    }
    // FIXME: Scoped non-blocking?
528
    butil::make_non_blocking(sockfd);
gejun's avatar
gejun committed
529 530 531 532
    const int rc = connect(sockfd, serv_addr, addrlen);
    if (rc == 0 || errno != EINPROGRESS) {
        return rc;
    }
533
#if defined(OS_LINUX)
gejun's avatar
gejun committed
534
    if (bthread_fd_wait(sockfd, EPOLLOUT) < 0) {
535 536 537
#elif defined(OS_MACOSX)
    if (bthread_fd_wait(sockfd, EVFILT_WRITE) < 0) {
#endif
gejun's avatar
gejun committed
538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554
        return -1;
    }
    int err;
    socklen_t errlen = sizeof(err);
    if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &err, &errlen) < 0) {
        PLOG(FATAL) << "Fail to getsockopt";
        return -1;
    }
    if (err != 0) {
        CHECK(err != EINPROGRESS);
        errno = err;
        return -1;
    }
    return 0;
}

// This does not wake pthreads calling bthread_fd_*wait.
gejun's avatar
gejun committed
555
int bthread_close(int fd) {
gejun's avatar
gejun committed
556 557 558 559
    return bthread::get_epoll_thread(fd).fd_close(fd);
}

}  // extern "C"