feature: only one event loop for all TcpTransport.

Author: ifplusor

src/transport/EventLoop.cpp

@@ -0,0 +1,109 @@
+/*
+ * 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.
+ */
+#include "EventLoop.h"
+#include <event2/thread.h>
+
+using namespace rocketmq;
+
+EventLoop *EventLoop::GetDefaultEventLoop() {
+  static EventLoop defaultEventLoop;
+  return &defaultEventLoop;
+}
+
+EventLoop::EventLoop(const struct event_config *config, bool run_immediately)
+    : m_eventBase(nullptr), m_loopThread(nullptr) {
+
+#ifdef WIN32
+  evthread_use_windows_threads();
+#else
+  evthread_use_pthreads();
+#endif
+
+  if (config == nullptr) {
+    m_eventBase = event_base_new();
+  } else {
+    m_eventBase = event_base_new_with_config(config);
+  }
+
+  if (m_eventBase == nullptr) {
+    // failure...
+  }
+
+  evthread_make_base_notifiable(m_eventBase);
+
+  if (run_immediately) {
+    start();
+  }
+}
+
+EventLoop::~EventLoop() {
+  stop();
+
+  if (m_eventBase != nullptr) {
+    event_base_free(m_eventBase);
+    m_eventBase = nullptr;
+  }
+}
+
+void EventLoop::start() {
+  if (m_loopThread == nullptr) {
+    // start event loop
+    m_loopThread = new boost::thread(boost::bind(&EventLoop::runLoop, this));
+  }
+}
+
+void EventLoop::stop() {
+  if (m_loopThread != nullptr) {
+    m_loopThread->interrupt();
+    m_loopThread->join();
+
+    delete m_loopThread;
+    m_loopThread = nullptr;
+  }
+}
+
+void EventLoop::runLoop() {
+  for (;;) {
+    int ret;
+
+    {
+      boost::this_thread::disable_interruption di;
+      ret = event_base_loop(m_eventBase, EVLOOP_NONBLOCK);
+    }
+
+    if (ret == 1) {
+      // no event
+      boost::this_thread::sleep(boost::posix_time::milliseconds(1));
+    }
+  }
+}
+
+BufferEvent *EventLoop::createBufferEvent(socket_t fd, int options) {
+  struct bufferevent *event = bufferevent_socket_new(m_eventBase, fd, options);
+  if (event == nullptr) {
+    return nullptr;
+  }
+
+  return new BufferEvent(event);
+}
+
+BufferEvent::~BufferEvent() {
+  if (m_bufferEvent != nullptr) {
+    bufferevent_free(m_bufferEvent);
+    m_bufferEvent = nullptr;
+  }
+}

src/transport/EventLoop.h

@@ -0,0 +1,94 @@
+/*
+ * 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.
+ */
+#ifndef __EVENTLOOP_H__
+#define __EVENTLOOP_H__
+
+#include <boost/noncopyable.hpp>
+#include <boost/thread/thread.hpp>
+
+#include <event2/buffer.h>
+#include <event2/event.h>
+#include <event2/bufferevent.h>
+
+#define socket_t evutil_socket_t
+
+namespace rocketmq {
+
+class BufferEvent;
+
+class EventLoop : public boost::noncopyable {
+ public:
+  static EventLoop *GetDefaultEventLoop();
+
+ public:
+  explicit EventLoop(const struct event_config *config = nullptr, bool run_immediately = true);
+  virtual ~EventLoop();
+
+  void start();
+  void stop();
+
+  BufferEvent *createBufferEvent(socket_t fd, int options);
+
+ private:
+  void runLoop();
+
+ private:
+  struct event_base *m_eventBase;
+  boost::thread *m_loopThread;
+};
+
+class BufferEvent : public boost::noncopyable {
+ public:
+  virtual ~BufferEvent();
+
+  void setCallback(bufferevent_data_cb readcb, bufferevent_data_cb writecb,
+                   bufferevent_event_cb eventcb, void *cbarg) {
+    // TODO: wrap callback
+    bufferevent_setcb(m_bufferEvent, readcb, writecb, eventcb, cbarg);
+  }
+
+  void setWatermark(short events, size_t lowmark, size_t highmark) {
+    bufferevent_setwatermark(m_bufferEvent, events, lowmark, highmark);
+  }
+
+  int enable(short event) {
+    return bufferevent_enable(m_bufferEvent, event);
+  }
+
+  int connect(const struct sockaddr *addr, int socklen) {
+    return bufferevent_socket_connect(m_bufferEvent, (struct sockaddr*) addr, socklen);
+  }
+
+  int write(const void *data, size_t size) {
+    return bufferevent_write(m_bufferEvent, data, size);
+  }
+
+  socket_t getfd() {
+    return bufferevent_getfd(m_bufferEvent);
+  }
+
+ private:
+  explicit BufferEvent(struct bufferevent *event) : m_bufferEvent(event) {};
+  friend EventLoop;
+
+ private:
+  struct bufferevent *m_bufferEvent;
+};
+
+}
+
+#endif //__EVENTLOOP_H__