1 // Copyright (c) 2011 The Chromium Authors. All rights reserved. 2 // Use of this source code is governed by a BSD-style license that can be 3 // found in the LICENSE file. 4 5 #include "base/message_pump_libevent.h" 6 7 #include <errno.h> 8 #include <fcntl.h> 9 10 #include "base/auto_reset.h" 11 #include "base/eintr_wrapper.h" 12 #include "base/logging.h" 13 #include "base/mac/scoped_nsautorelease_pool.h" 14 #include "base/memory/scoped_ptr.h" 15 #include "base/observer_list.h" 16 #include "base/time.h" 17 #if defined(USE_SYSTEM_LIBEVENT) 18 #include <event.h> 19 #else 20 #include "third_party/libevent/event.h" 21 #endif 22 23 // Lifecycle of struct event 24 // Libevent uses two main data structures: 25 // struct event_base (of which there is one per message pump), and 26 // struct event (of which there is roughly one per socket). 27 // The socket's struct event is created in 28 // MessagePumpLibevent::WatchFileDescriptor(), 29 // is owned by the FileDescriptorWatcher, and is destroyed in 30 // StopWatchingFileDescriptor(). 31 // It is moved into and out of lists in struct event_base by 32 // the libevent functions event_add() and event_del(). 33 // 34 // TODO(dkegel): 35 // At the moment bad things happen if a FileDescriptorWatcher 36 // is active after its MessagePumpLibevent has been destroyed. 37 // See MessageLoopTest.FileDescriptorWatcherOutlivesMessageLoop 38 // Not clear yet whether that situation occurs in practice, 39 // but if it does, we need to fix it. 40 41 namespace base { 42 43 // Return 0 on success 44 // Too small a function to bother putting in a library? 45 static int SetNonBlocking(int fd) { 46 int flags = fcntl(fd, F_GETFL, 0); 47 if (flags == -1) 48 flags = 0; 49 return fcntl(fd, F_SETFL, flags | O_NONBLOCK); 50 } 51 52 MessagePumpLibevent::FileDescriptorWatcher::FileDescriptorWatcher() 53 : is_persistent_(false), 54 event_(NULL), 55 pump_(NULL), 56 watcher_(NULL) { 57 } 58 59 MessagePumpLibevent::FileDescriptorWatcher::~FileDescriptorWatcher() { 60 if (event_) { 61 StopWatchingFileDescriptor(); 62 } 63 } 64 65 bool MessagePumpLibevent::FileDescriptorWatcher::StopWatchingFileDescriptor() { 66 event* e = ReleaseEvent(); 67 if (e == NULL) 68 return true; 69 70 // event_del() is a no-op if the event isn't active. 71 int rv = event_del(e); 72 delete e; 73 pump_ = NULL; 74 watcher_ = NULL; 75 return (rv == 0); 76 } 77 78 void MessagePumpLibevent::FileDescriptorWatcher::Init(event *e, 79 bool is_persistent) { 80 DCHECK(e); 81 DCHECK(!event_); 82 83 is_persistent_ = is_persistent; 84 event_ = e; 85 } 86 87 event *MessagePumpLibevent::FileDescriptorWatcher::ReleaseEvent() { 88 struct event *e = event_; 89 event_ = NULL; 90 return e; 91 } 92 93 void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanReadWithoutBlocking( 94 int fd, MessagePumpLibevent* pump) { 95 pump->WillProcessIOEvent(); 96 watcher_->OnFileCanReadWithoutBlocking(fd); 97 pump->DidProcessIOEvent(); 98 } 99 100 void MessagePumpLibevent::FileDescriptorWatcher::OnFileCanWriteWithoutBlocking( 101 int fd, MessagePumpLibevent* pump) { 102 pump->WillProcessIOEvent(); 103 watcher_->OnFileCanWriteWithoutBlocking(fd); 104 pump->DidProcessIOEvent(); 105 } 106 107 MessagePumpLibevent::MessagePumpLibevent() 108 : keep_running_(true), 109 in_run_(false), 110 event_base_(event_base_new()), 111 wakeup_pipe_in_(-1), 112 wakeup_pipe_out_(-1) { 113 if (!Init()) 114 NOTREACHED(); 115 } 116 117 MessagePumpLibevent::~MessagePumpLibevent() { 118 DCHECK(wakeup_event_); 119 DCHECK(event_base_); 120 event_del(wakeup_event_); 121 delete wakeup_event_; 122 if (wakeup_pipe_in_ >= 0) { 123 if (HANDLE_EINTR(close(wakeup_pipe_in_)) < 0) 124 PLOG(ERROR) << "close"; 125 } 126 if (wakeup_pipe_out_ >= 0) { 127 if (HANDLE_EINTR(close(wakeup_pipe_out_)) < 0) 128 PLOG(ERROR) << "close"; 129 } 130 event_base_free(event_base_); 131 } 132 133 bool MessagePumpLibevent::WatchFileDescriptor(int fd, 134 bool persistent, 135 Mode mode, 136 FileDescriptorWatcher *controller, 137 Watcher *delegate) { 138 DCHECK_GE(fd, 0); 139 DCHECK(controller); 140 DCHECK(delegate); 141 DCHECK(mode == WATCH_READ || mode == WATCH_WRITE || mode == WATCH_READ_WRITE); 142 143 int event_mask = persistent ? EV_PERSIST : 0; 144 if ((mode & WATCH_READ) != 0) { 145 event_mask |= EV_READ; 146 } 147 if ((mode & WATCH_WRITE) != 0) { 148 event_mask |= EV_WRITE; 149 } 150 151 scoped_ptr<event> evt(controller->ReleaseEvent()); 152 if (evt.get() == NULL) { 153 // Ownership is transferred to the controller. 154 evt.reset(new event); 155 } else { 156 // Make sure we don't pick up any funky internal libevent masks. 157 int old_interest_mask = evt.get()->ev_events & 158 (EV_READ | EV_WRITE | EV_PERSIST); 159 160 // Combine old/new event masks. 161 event_mask |= old_interest_mask; 162 163 // Must disarm the event before we can reuse it. 164 event_del(evt.get()); 165 166 // It's illegal to use this function to listen on 2 separate fds with the 167 // same |controller|. 168 if (EVENT_FD(evt.get()) != fd) { 169 NOTREACHED() << "FDs don't match" << EVENT_FD(evt.get()) << "!=" << fd; 170 return false; 171 } 172 } 173 174 // Set current interest mask and message pump for this event. 175 event_set(evt.get(), fd, event_mask, OnLibeventNotification, controller); 176 177 // Tell libevent which message pump this socket will belong to when we add it. 178 if (event_base_set(event_base_, evt.get()) != 0) { 179 return false; 180 } 181 182 // Add this socket to the list of monitored sockets. 183 if (event_add(evt.get(), NULL) != 0) { 184 return false; 185 } 186 187 // Transfer ownership of evt to controller. 188 controller->Init(evt.release(), persistent); 189 190 controller->set_watcher(delegate); 191 controller->set_pump(this); 192 193 return true; 194 } 195 196 void MessagePumpLibevent::AddIOObserver(IOObserver *obs) { 197 io_observers_.AddObserver(obs); 198 } 199 200 void MessagePumpLibevent::RemoveIOObserver(IOObserver *obs) { 201 io_observers_.RemoveObserver(obs); 202 } 203 204 // Tell libevent to break out of inner loop. 205 static void timer_callback(int fd, short events, void *context) 206 { 207 event_base_loopbreak((struct event_base *)context); 208 } 209 210 // Reentrant! 211 void MessagePumpLibevent::Run(Delegate* delegate) { 212 DCHECK(keep_running_) << "Quit must have been called outside of Run!"; 213 AutoReset<bool> auto_reset_in_run(&in_run_, true); 214 215 // event_base_loopexit() + EVLOOP_ONCE is leaky, see http://crbug.com/25641. 216 // Instead, make our own timer and reuse it on each call to event_base_loop(). 217 scoped_ptr<event> timer_event(new event); 218 219 for (;;) { 220 mac::ScopedNSAutoreleasePool autorelease_pool; 221 222 bool did_work = delegate->DoWork(); 223 if (!keep_running_) 224 break; 225 226 did_work |= delegate->DoDelayedWork(&delayed_work_time_); 227 if (!keep_running_) 228 break; 229 230 if (did_work) 231 continue; 232 233 did_work = delegate->DoIdleWork(); 234 if (!keep_running_) 235 break; 236 237 if (did_work) 238 continue; 239 240 // EVLOOP_ONCE tells libevent to only block once, 241 // but to service all pending events when it wakes up. 242 if (delayed_work_time_.is_null()) { 243 event_base_loop(event_base_, EVLOOP_ONCE); 244 } else { 245 TimeDelta delay = delayed_work_time_ - TimeTicks::Now(); 246 if (delay > TimeDelta()) { 247 struct timeval poll_tv; 248 poll_tv.tv_sec = delay.InSeconds(); 249 poll_tv.tv_usec = delay.InMicroseconds() % Time::kMicrosecondsPerSecond; 250 event_set(timer_event.get(), -1, 0, timer_callback, event_base_); 251 event_base_set(event_base_, timer_event.get()); 252 event_add(timer_event.get(), &poll_tv); 253 event_base_loop(event_base_, EVLOOP_ONCE); 254 event_del(timer_event.get()); 255 } else { 256 // It looks like delayed_work_time_ indicates a time in the past, so we 257 // need to call DoDelayedWork now. 258 delayed_work_time_ = TimeTicks(); 259 } 260 } 261 } 262 263 keep_running_ = true; 264 } 265 266 void MessagePumpLibevent::Quit() { 267 DCHECK(in_run_); 268 // Tell both libevent and Run that they should break out of their loops. 269 keep_running_ = false; 270 ScheduleWork(); 271 } 272 273 void MessagePumpLibevent::ScheduleWork() { 274 // Tell libevent (in a threadsafe way) that it should break out of its loop. 275 char buf = 0; 276 int nwrite = HANDLE_EINTR(write(wakeup_pipe_in_, &buf, 1)); 277 DCHECK(nwrite == 1 || errno == EAGAIN) 278 << "[nwrite:" << nwrite << "] [errno:" << errno << "]"; 279 } 280 281 void MessagePumpLibevent::ScheduleDelayedWork( 282 const TimeTicks& delayed_work_time) { 283 // We know that we can't be blocked on Wait right now since this method can 284 // only be called on the same thread as Run, so we only need to update our 285 // record of how long to sleep when we do sleep. 286 delayed_work_time_ = delayed_work_time; 287 } 288 289 void MessagePumpLibevent::WillProcessIOEvent() { 290 FOR_EACH_OBSERVER(IOObserver, io_observers_, WillProcessIOEvent()); 291 } 292 293 void MessagePumpLibevent::DidProcessIOEvent() { 294 FOR_EACH_OBSERVER(IOObserver, io_observers_, DidProcessIOEvent()); 295 } 296 297 bool MessagePumpLibevent::Init() { 298 int fds[2]; 299 if (pipe(fds)) { 300 DLOG(ERROR) << "pipe() failed, errno: " << errno; 301 return false; 302 } 303 if (SetNonBlocking(fds[0])) { 304 DLOG(ERROR) << "SetNonBlocking for pipe fd[0] failed, errno: " << errno; 305 return false; 306 } 307 if (SetNonBlocking(fds[1])) { 308 DLOG(ERROR) << "SetNonBlocking for pipe fd[1] failed, errno: " << errno; 309 return false; 310 } 311 wakeup_pipe_out_ = fds[0]; 312 wakeup_pipe_in_ = fds[1]; 313 314 wakeup_event_ = new event; 315 event_set(wakeup_event_, wakeup_pipe_out_, EV_READ | EV_PERSIST, 316 OnWakeup, this); 317 event_base_set(event_base_, wakeup_event_); 318 319 if (event_add(wakeup_event_, 0)) 320 return false; 321 return true; 322 } 323 324 // static 325 void MessagePumpLibevent::OnLibeventNotification(int fd, short flags, 326 void* context) { 327 FileDescriptorWatcher* controller = 328 static_cast<FileDescriptorWatcher*>(context); 329 330 MessagePumpLibevent* pump = controller->pump(); 331 332 if (flags & EV_WRITE) { 333 controller->OnFileCanWriteWithoutBlocking(fd, pump); 334 } 335 if (flags & EV_READ) { 336 controller->OnFileCanReadWithoutBlocking(fd, pump); 337 } 338 } 339 340 // Called if a byte is received on the wakeup pipe. 341 // static 342 void MessagePumpLibevent::OnWakeup(int socket, short flags, void* context) { 343 base::MessagePumpLibevent* that = 344 static_cast<base::MessagePumpLibevent*>(context); 345 DCHECK(that->wakeup_pipe_out_ == socket); 346 347 // Remove and discard the wakeup byte. 348 char buf; 349 int nread = HANDLE_EINTR(read(socket, &buf, 1)); 350 DCHECK_EQ(nread, 1); 351 // Tell libevent to break out of inner loop. 352 event_base_loopbreak(that->event_base_); 353 } 354 355 } // namespace base 356