| 420 | } |
| 421 | |
| 422 | int EventCenter::process_events(unsigned timeout_microseconds, ceph::timespan *working_dur) |
| 423 | { |
| 424 | struct timeval tv; |
| 425 | int numevents; |
| 426 | bool trigger_time = false; |
| 427 | auto now = clock_type::now(); |
| 428 | clock_type::time_point end_time = now + std::chrono::microseconds(timeout_microseconds); |
| 429 | |
| 430 | auto it = time_events.begin(); |
| 431 | if (it != time_events.end() && end_time >= it->first) { |
| 432 | trigger_time = true; |
| 433 | end_time = it->first; |
| 434 | |
| 435 | if (end_time > now) { |
| 436 | timeout_microseconds = std::chrono::duration_cast<std::chrono::microseconds>(end_time - now).count(); |
| 437 | timeout_microseconds = std::max<unsigned>(timeout_microseconds, |
| 438 | cct->_conf->ms_time_events_min_wait_interval); |
| 439 | } else { |
| 440 | timeout_microseconds = 0; |
| 441 | } |
| 442 | } |
| 443 | |
| 444 | bool blocking = pollers.empty() && !external_num_events.load(); |
| 445 | if (!blocking) |
| 446 | timeout_microseconds = 0; |
| 447 | tv.tv_sec = timeout_microseconds / 1000000; |
| 448 | tv.tv_usec = timeout_microseconds % 1000000; |
| 449 | |
| 450 | ldout(cct, 30) << __func__ << " wait second " << tv.tv_sec << " usec " << tv.tv_usec << dendl; |
| 451 | std::vector<FiredFileEvent> fired_events; |
| 452 | numevents = driver->event_wait(fired_events, &tv); |
| 453 | auto working_start = ceph::mono_clock::now(); |
| 454 | for (int event_id = 0; event_id < numevents; event_id++) { |
| 455 | int rfired = 0; |
| 456 | FileEvent *event; |
| 457 | EventCallbackRef cb; |
| 458 | event = _get_file_event(fired_events[event_id].fd); |
| 459 | |
| 460 | /* note the event->mask & mask & ... code: maybe an already processed |
| 461 | * event removed an element that fired and we still didn't |
| 462 | * processed, so we check if the event is still valid. */ |
| 463 | if (event->mask & fired_events[event_id].mask & EVENT_READABLE) { |
| 464 | rfired = 1; |
| 465 | cb = event->read_cb; |
| 466 | cb->do_request(fired_events[event_id].fd); |
| 467 | } |
| 468 | |
| 469 | if (event->mask & fired_events[event_id].mask & EVENT_WRITABLE) { |
| 470 | if (!rfired || event->read_cb != event->write_cb) { |
| 471 | cb = event->write_cb; |
| 472 | cb->do_request(fired_events[event_id].fd); |
| 473 | } |
| 474 | } |
| 475 | |
| 476 | ldout(cct, 30) << __func__ << " event_wq process is " << fired_events[event_id].fd |
| 477 | << " mask is " << fired_events[event_id].mask << dendl; |
| 478 | } |
| 479 | |