fd.cpp 17.9 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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
18 19 20 21 22
// bthread - A M:N threading library to make applications more concurrent.

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

gejun's avatar
gejun committed
23
#include "butil/compat.h"
gejun's avatar
gejun committed
24 25
#include <new>                                   // std::nothrow
#include <sys/poll.h>                            // poll()
zhujiashun's avatar
zhujiashun committed
26
#if defined(OS_MACOSX)
27 28
#include <sys/types.h>                           // struct kevent
#include <sys/event.h>                           // kevent(), kqueue()
zhujiashun's avatar
zhujiashun committed
29
#endif
30 31 32 33 34
#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
35 36 37 38 39 40 41 42 43 44 45 46 47
#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 {
48
        butil::atomic<T> items[BLOCK_SIZE];
gejun's avatar
gejun committed
49 50 51 52
    };

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

56
    butil::atomic<T>* get_or_new(size_t index) {
gejun's avatar
gejun committed
57 58 59 60 61
        const size_t block_index = index / BLOCK_SIZE;
        if (block_index >= NBLOCK) {
            return NULL;
        }
        const size_t block_offset = index - block_index * BLOCK_SIZE;
62
        Block* b = _blocks[block_index].load(butil::memory_order_consume);
gejun's avatar
gejun committed
63 64 65 66 67
        if (b != NULL) {
            return b->items + block_offset;
        }
        b = new (std::nothrow) Block;
        if (NULL == b) {
68
            b = _blocks[block_index].load(butil::memory_order_consume);
gejun's avatar
gejun committed
69 70 71 72 73 74
            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(
75 76
                expected, b, butil::memory_order_release,
                butil::memory_order_consume)) {
gejun's avatar
gejun committed
77 78 79 80 81 82
            return b->items + block_offset;
        }
        delete b;
        return expected->items + block_offset;
    }

83
    butil::atomic<T>* get(size_t index) const {
gejun's avatar
gejun committed
84 85 86
        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;
87
            Block* const b = _blocks[block_index].load(butil::memory_order_consume);
gejun's avatar
gejun committed
88 89 90 91 92 93 94 95
            if (__builtin_expect(b != NULL, 1)) {
                return b->items + block_offset;
            }
        }
        return NULL;
    }

private:
96
    butil::atomic<Block*> _blocks[NBLOCK];
gejun's avatar
gejun committed
97 98
};

99
typedef butil::atomic<int> EpollButex;
gejun's avatar
gejun committed
100 101 102 103

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

#ifndef NDEBUG
104
butil::static_atomic<int> break_nums = BUTIL_STATIC_ATOMIC_INIT(0);
gejun's avatar
gejun committed
105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123
#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;
        }
124 125 126 127 128 129
        _start_mutex.lock();
        // Double check
        if (started()) {
            _start_mutex.unlock();
            return -1;
        }
130
#if defined(OS_LINUX)
gejun's avatar
gejun committed
131
        _epfd = epoll_create(epoll_size);
132 133 134
#elif defined(OS_MACOSX)
        _epfd = kqueue();
#endif
135
        _start_mutex.unlock();
gejun's avatar
gejun committed
136
        if (_epfd < 0) {
137
            PLOG(FATAL) << "Fail to epoll_create/kqueue";
gejun's avatar
gejun committed
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 170 171 172 173
            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;
        }
174
#if defined(OS_LINUX)
gejun's avatar
gejun committed
175 176 177
        epoll_event evt = { EPOLLOUT, { NULL } };
        if (epoll_ctl(saved_epfd, EPOLL_CTL_ADD,
                      closing_epoll_pipe[1], &evt) < 0) {
178 179 180 181 182 183
#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
184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199
            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;
    }

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

207
        EpollButex* butex = p->load(butil::memory_order_consume);
gejun's avatar
gejun committed
208 209 210 211 212
        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>();
213
            butex->store(0, butil::memory_order_relaxed);
gejun's avatar
gejun committed
214 215
            EpollButex* expected = NULL;
            if (!p->compare_exchange_strong(expected, butex,
216 217
                                            butil::memory_order_release,
                                            butil::memory_order_consume)) {
gejun's avatar
gejun committed
218 219 220 221 222 223 224 225 226
                butex_destroy(butex);
                butex = expected;
            }
        }
        
        while (butex == CLOSING_GUARD) {  // bthread_close() is running.
            if (sched_yield() < 0) {
                return -1;
            }
227
            butex = p->load(butil::memory_order_consume);
gejun's avatar
gejun committed
228 229 230 231
        }
        // 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.
232
        const int expected_val = butex->load(butil::memory_order_relaxed);
gejun's avatar
gejun committed
233

234 235 236
#if defined(OS_LINUX)
# ifdef BAIDU_KERNEL_FIXED_EPOLLONESHOT_BUG
        epoll_event evt = { events | EPOLLONESHOT, { butex } };
gejun's avatar
gejun committed
237 238 239 240 241 242 243
        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;
            }
        }
244
# else
gejun's avatar
gejun committed
245
        epoll_event evt;
246
        evt.events = events;
gejun's avatar
gejun committed
247 248 249 250 251 252
        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;
        }
253 254 255 256 257 258 259 260 261 262
# 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
263 264 265
        if (butex_wait(butex, expected_val, abstime) < 0 &&
            errno != EWOULDBLOCK && errno != EINTR) {
            return -1;
gejun's avatar
gejun committed
266
        }
267
        return 0;
gejun's avatar
gejun committed
268 269 270 271 272 273 274 275
    }

    int fd_close(int fd) {
        if (fd < 0) {
            // what close(-1) returns
            errno = EBADF;
            return -1;
        }
276
        butil::atomic<EpollButex*>* pbutex = bthread::fd_butexes.get(fd);
gejun's avatar
gejun committed
277 278 279 280 281
        if (NULL == pbutex) {
            // Did not call bthread_fd functions, close directly.
            return close(fd);
        }
        EpollButex* butex = pbutex->exchange(
282
            CLOSING_GUARD, butil::memory_order_relaxed);
gejun's avatar
gejun committed
283 284 285 286 287 288
        if (butex == CLOSING_GUARD) {
            // concurrent double close detected.
            errno = EBADF;
            return -1;
        }
        if (butex != NULL) {
289
            butex->fetch_add(1, butil::memory_order_relaxed);
gejun's avatar
gejun committed
290 291
            butex_wake_all(butex);
        }
292
#if defined(OS_LINUX)
gejun's avatar
gejun committed
293
        epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, NULL);
294 295 296 297 298 299 300
#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
301
        const int rc = close(fd);
302
        pbutex->exchange(butex, butil::memory_order_relaxed);
gejun's avatar
gejun committed
303 304 305 306 307 308 309 310 311 312 313 314 315 316 317
        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;
318
#if defined(OS_LINUX)
gejun's avatar
gejun committed
319
        epoll_event* e = new (std::nothrow) epoll_event[MAX_EVENTS];
320 321 322 323
#elif defined(OS_MACOSX)
        typedef struct kevent KEVENT;
        struct kevent* e = new (std::nothrow) KEVENT[MAX_EVENTS];
#endif
gejun's avatar
gejun committed
324 325 326 327 328
        if (NULL == e) {
            LOG(FATAL) << "Fail to new epoll_event";
            return NULL;
        }

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

            if (n < 0) {
                if (errno == EINTR) {
#ifndef NDEBUG
348
                    break_nums.fetch_add(1, butil::memory_order_relaxed);
gejun's avatar
gejun committed
349 350 351 352 353 354 355 356 357 358 359 360 361
                    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;
            }

362 363
#if defined(OS_LINUX)
# ifndef BAIDU_KERNEL_FIXED_EPOLLONESHOT_BUG
gejun's avatar
gejun committed
364 365 366
            for (int i = 0; i < n; ++i) {
                epoll_ctl(epfd, EPOLL_CTL_DEL, e[i].data.fd, NULL);
            }
367
# endif
gejun's avatar
gejun committed
368 369
#endif
            for (int i = 0; i < n; ++i) {
370 371
#if defined(OS_LINUX)
# ifdef BAIDU_KERNEL_FIXED_EPOLLONESHOT_BUG
gejun's avatar
gejun committed
372
                EpollButex* butex = static_cast<EpollButex*>(e[i].data.ptr);
373
# else
374
                butil::atomic<EpollButex*>* pbutex = fd_butexes.get(e[i].data.fd);
gejun's avatar
gejun committed
375
                EpollButex* butex = pbutex ?
376
                    pbutex->load(butil::memory_order_consume) : NULL;
377 378 379
# endif
#elif defined(OS_MACOSX)
                EpollButex* butex = static_cast<EpollButex*>(e[i].udata);
gejun's avatar
gejun committed
380 381
#endif
                if (butex != NULL && butex != CLOSING_GUARD) {
382
                    butex->fetch_add(1, butil::memory_order_relaxed);
gejun's avatar
gejun committed
383 384 385 386 387 388 389 390 391 392 393 394 395 396
                    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;
397
    butil::Mutex _start_mutex;
gejun's avatar
gejun committed
398 399 400 401 402 403 404 405 406 407 408
};

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;
    }

409
    EpollThread& et = epoll_thread[butil::fmix32(fd) % BTHREAD_EPOLL_THREAD_NUM];
gejun's avatar
gejun committed
410 411 412 413
    et.start(BTHREAD_DEFAULT_EPOLL_SIZE);
    return et;
}

414
//TODO(zhujiashun): change name
gejun's avatar
gejun committed
415 416 417 418 419 420 421 422 423 424 425
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;
}

426
#if defined(OS_LINUX)
gejun's avatar
gejun committed
427 428 429 430 431 432 433 434 435 436
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;
}
437 438 439 440 441 442 443 444 445 446 447 448 449
#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
450 451

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

495
int bthread_fd_wait(int fd, unsigned events) {
gejun's avatar
gejun committed
496 497 498 499 500 501 502
    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(
503
            fd, events, NULL);
gejun's avatar
gejun committed
504
    }
505
    return bthread::pthread_fd_wait(fd, events, NULL);
gejun's avatar
gejun committed
506 507
}

508
int bthread_fd_timedwait(int fd, unsigned events,
gejun's avatar
gejun committed
509
                         const timespec* abstime) {
gejun's avatar
gejun committed
510
    if (NULL == abstime) {
511
        return bthread_fd_wait(fd, events);
gejun's avatar
gejun committed
512 513 514 515 516 517 518 519
    }
    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(
520
            fd, events, abstime);
gejun's avatar
gejun committed
521
    }
522
    return bthread::pthread_fd_wait(fd, events, abstime);
gejun's avatar
gejun committed
523 524 525
}

int bthread_connect(int sockfd, const sockaddr* serv_addr,
gejun's avatar
gejun committed
526
                    socklen_t addrlen) {
gejun's avatar
gejun committed
527 528 529 530 531
    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?
532
    butil::make_non_blocking(sockfd);
gejun's avatar
gejun committed
533 534 535 536
    const int rc = connect(sockfd, serv_addr, addrlen);
    if (rc == 0 || errno != EINPROGRESS) {
        return rc;
    }
537
#if defined(OS_LINUX)
gejun's avatar
gejun committed
538
    if (bthread_fd_wait(sockfd, EPOLLOUT) < 0) {
539 540 541
#elif defined(OS_MACOSX)
    if (bthread_fd_wait(sockfd, EVFILT_WRITE) < 0) {
#endif
gejun's avatar
gejun committed
542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558
        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
559
int bthread_close(int fd) {
gejun's avatar
gejun committed
560 561 562 563
    return bthread::get_epoll_thread(fd).fd_close(fd);
}

}  // extern "C"