You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafficserver.apache.org by sh...@apache.org on 2018/03/19 20:11:02 UTC
[trafficserver] branch master updated: remove traffic_ctl delay
This is an automated email from the ASF dual-hosted git repository.
shinrich pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/trafficserver.git
The following commit(s) were added to refs/heads/master by this push:
new d7a6c42 remove traffic_ctl delay
d7a6c42 is described below
commit d7a6c42c7302f523d2abf0c4ca8cd8b30f449624
Author: Alan Wang <xf...@gmail.com>
AuthorDate: Thu Mar 8 21:35:55 2018 -0800
remove traffic_ctl delay
---
mgmt/BaseManager.cc | 38 -----------
mgmt/BaseManager.h | 3 -
mgmt/LocalManager.cc | 58 ++++++++++++++++-
mgmt/LocalManager.h | 10 ++-
mgmt/ProcessManager.cc | 169 +++++++++++++++++++++++++++++++------------------
mgmt/ProcessManager.h | 7 ++
6 files changed, 178 insertions(+), 107 deletions(-)
diff --git a/mgmt/BaseManager.cc b/mgmt/BaseManager.cc
index 7dab2aa..e2209f7 100644
--- a/mgmt/BaseManager.cc
+++ b/mgmt/BaseManager.cc
@@ -113,44 +113,6 @@ BaseManager::registerMgmtCallback(int msg_id, MgmtCallback cb, void *opaque_cb_d
return msg_id;
} /* End BaseManager::registerMgmtCallback */
-/*
- * signalMgmtEntity(...)
- */
-int
-BaseManager::signalMgmtEntity(int msg_id, char *data_str)
-{
- if (data_str) {
- return signalMgmtEntity(msg_id, data_str, strlen(data_str) + 1);
- } else {
- return signalMgmtEntity(msg_id, nullptr, 0);
- }
-
-} /* End BaseManager::signalMgmtEntity */
-
-/*
- * signalMgmtEntity(...)
- */
-int
-BaseManager::signalMgmtEntity(int msg_id, char *data_raw, int data_len)
-{
- MgmtMessageHdr *mh;
-
- if (data_raw) {
- mh = (MgmtMessageHdr *)ats_malloc(sizeof(MgmtMessageHdr) + data_len);
- mh->msg_id = msg_id;
- mh->data_len = data_len;
- memcpy((char *)mh + sizeof(MgmtMessageHdr), data_raw, data_len);
- } else {
- mh = (MgmtMessageHdr *)ats_malloc(sizeof(MgmtMessageHdr));
- mh->msg_id = msg_id;
- mh->data_len = 0;
- }
-
- ink_release_assert(enqueue(mgmt_event_queue, mh));
- return msg_id;
-
-} /* End BaseManager::signalMgmtEntity */
-
void
BaseManager::executeMgmtCallback(int msg_id, char *data_raw, int data_len)
{
diff --git a/mgmt/BaseManager.h b/mgmt/BaseManager.h
index f57df76..c71be3d 100644
--- a/mgmt/BaseManager.h
+++ b/mgmt/BaseManager.h
@@ -124,9 +124,6 @@ public:
int registerMgmtCallback(int msg_id, MgmtCallback func, void *opaque_callback_data = NULL);
- int signalMgmtEntity(int msg_id, char *data_str = NULL);
- int signalMgmtEntity(int msg_id, char *data_raw, int data_len);
-
LLQ *mgmt_event_queue;
InkHashTable *mgmt_callback_table;
diff --git a/mgmt/LocalManager.cc b/mgmt/LocalManager.cc
index 05abb18..c4a08c0 100644
--- a/mgmt/LocalManager.cc
+++ b/mgmt/LocalManager.cc
@@ -45,6 +45,13 @@ LocalManager::mgmtCleanup()
close_socket(process_server_sockfd);
process_server_sockfd = ts::NO_FD;
+#if HAVE_EVENTFD
+ if (wakeup_fd != ts::NO_FD) {
+ close_socket(wakeup_fd);
+ wakeup_fd = ts::NO_FD;
+ }
+#endif
+
// fix me for librecords
closelog();
@@ -265,6 +272,13 @@ LocalManager::initMgmtProcessServer()
mgmt_fatal(errno, "[LocalManager::initMgmtProcessServer] failed to bind socket at %s\n", sockpath.c_str());
}
+#if HAVE_EVENTFD
+ wakeup_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
+ if (wakeup_fd < 0) {
+ mgmt_fatal(errno, "[LocalManager::initMgmtProcessServer] failed to create eventfd. errno : %s\n", strerror(errno));
+ }
+#endif
+
umask(oldmask);
RecSetRecordInt("proxy.node.restarts.manager.start_time", manager_started_at, REC_SOURCE_DEFAULT);
}
@@ -314,6 +328,12 @@ LocalManager::pollMgmtProcessServer()
}
#endif
+#if HAVE_EVENTFD
+ if (wakeup_fd != ts::NO_FD) {
+ FD_SET(wakeup_fd, &fdlist);
+ }
+#endif
+
num = mgmt_select(FD_SETSIZE, &fdlist, nullptr, nullptr, &timeout);
switch (num) {
@@ -329,11 +349,14 @@ LocalManager::pollMgmtProcessServer()
return;
default:
-
+ // if we get a wakeup_fd event, we may not want to follow it
+ // because there may be more data to be read on the socket.
+ bool keep_polling = false;
#if TS_HAS_WCCP
if (wccp_fd != ts::NO_FD && FD_ISSET(wccp_fd, &fdlist)) {
wccp_cache.handleMessage();
--num;
+ keep_polling = true;
}
#endif
@@ -352,6 +375,7 @@ LocalManager::pollMgmtProcessServer()
close_socket(new_sockfd);
}
--num;
+ keep_polling = true;
}
if (ts::NO_FD != watched_process_fd && FD_ISSET(watched_process_fd, &fdlist)) {
@@ -360,6 +384,8 @@ LocalManager::pollMgmtProcessServer()
MgmtMessageHdr *mh_full;
char *data_raw;
+ keep_polling = true;
+
// read the message
if ((res = mgmt_read_pipe(watched_process_fd, (char *)&mh_hdr, sizeof(MgmtMessageHdr))) > 0) {
mh_full = (MgmtMessageHdr *)alloca(sizeof(MgmtMessageHdr) + mh_hdr.data_len);
@@ -413,9 +439,23 @@ LocalManager::pollMgmtProcessServer()
RecSetRecordInt("proxy.node.proxy_running", 0, REC_SOURCE_DEFAULT);
}
- num--;
+ --num;
}
+#if HAVE_EVENTFD
+ if (wakeup_fd != ts::NO_FD && FD_ISSET(wakeup_fd, &fdlist)) {
+ if (!keep_polling) {
+ // read or else fd will always be set.
+ uint64_t ignore;
+ ATS_UNUSED_RETURN(read(wakeup_fd, &ignore, sizeof(uint64_t)));
+ return;
+ }
+ --num;
+ }
+#else
+ (void)keep_polling; // suppress compiler warning
+#endif
+
ink_assert(num == 0); /* Invariant */
}
}
@@ -701,7 +741,19 @@ LocalManager::signalEvent(int msg_id, const char *data_raw, int data_len)
memcpy((char *)mh + sizeof(MgmtMessageHdr), data_raw, data_len);
ink_assert(enqueue(mgmt_event_queue, mh));
- return;
+#if HAVE_EVENTFD
+ // we don't care about the actual value of wakeup_fd, so just keep adding 1. just need to
+ // wakeup the fd. also, note that wakeup_fd was initalized to non-blocking so we can
+ // directly write to it without any timeout checking.
+ //
+ // don't tigger if MGMT_EVENT_LIBRECORD because they happen all the time
+ // and don't require a quick response. for MGMT_EVENT_LIBRECORD, rely on timeouts so
+ // traffic_server can spend more time doing other things
+ uint64_t one = 1;
+ if (wakeup_fd != ts::NO_FD && mh->msg_id != MGMT_EVENT_LIBRECORDS) {
+ ATS_UNUSED_RETURN(write(wakeup_fd, &one, sizeof(uint64_t))); // trigger to stop polling
+ }
+#endif
}
/*
diff --git a/mgmt/LocalManager.h b/mgmt/LocalManager.h
index 524b086..082e134 100644
--- a/mgmt/LocalManager.h
+++ b/mgmt/LocalManager.h
@@ -37,10 +37,13 @@
#include "Alarms.h"
#include "BaseManager.h"
#include <records/I_RecHttp.h>
+#include <syslog.h>
#if TS_HAS_WCCP
#include <wccp/Wccp.h>
#endif
-#include <syslog.h>
+#if HAVE_EVENTFD
+#include <sys/eventfd.h>
+#endif
class FileManager;
@@ -120,7 +123,10 @@ public:
int process_server_sockfd = ts::NO_FD;
int watched_process_fd = ts::NO_FD;
- pid_t proxy_launch_pid = -1;
+#if HAVE_EVENTFD
+ int wakeup_fd = ts::NO_FD; // external trigger to stop polling
+#endif
+ pid_t proxy_launch_pid = -1;
Alarms *alarm_keeper = nullptr;
FileManager *configFiles = nullptr;
diff --git a/mgmt/ProcessManager.cc b/mgmt/ProcessManager.cc
index 7122838..7fd5cc8 100644
--- a/mgmt/ProcessManager.cc
+++ b/mgmt/ProcessManager.cc
@@ -40,21 +40,13 @@ inkcoreapi ProcessManager *pmgmt = nullptr;
// socket. Returns -errno on error, otherwise 0. If a message was read the
// *msg pointer will be filled in with the message that was read.
static int
-read_management_message(int sockfd, ink_hrtime timeout, MgmtMessageHdr **msg)
+read_management_message(int sockfd, MgmtMessageHdr **msg)
{
MgmtMessageHdr hdr;
int ret;
*msg = nullptr;
- switch (mgmt_read_timeout(sockfd, ink_hrtime_to_sec(timeout), 0 /* usec */)) {
- case 0:
- // Timed out.
- return 0;
- case -1:
- return -errno;
- }
-
// We have a message, try to read the message header.
ret = mgmt_read_pipe(sockfd, reinterpret_cast<char *>(&hdr), sizeof(MgmtMessageHdr));
switch (ret) {
@@ -65,7 +57,7 @@ read_management_message(int sockfd, ink_hrtime timeout, MgmtMessageHdr **msg)
break;
default:
// Received -errno.
- return ret;
+ return -errno;
}
size_t msg_size = sizeof(MgmtMessageHdr) + hdr.data_len;
@@ -111,10 +103,22 @@ ProcessManager::stop()
ink_release_assert(running == 1);
ink_atomic_decrement(&running, 1);
- int tmp = local_manager_sockfd;
+ int tmp;
+
+ if (local_manager_sockfd != ts::NO_FD) {
+ tmp = local_manager_sockfd;
+ local_manager_sockfd = ts::NO_FD;
+ close_socket(tmp);
+ }
+
+#if HAVE_EVENTFD
+ if (wakeup_fd != ts::NO_FD) {
+ tmp = wakeup_fd;
+ wakeup_fd = ts::NO_FD;
+ close_socket(tmp);
+ }
+#endif
- local_manager_sockfd = -1;
- close_socket(tmp);
ink_thread_kill(poll_thread, SIGINT);
ink_thread_join(poll_thread);
@@ -166,13 +170,10 @@ ProcessManager::processManagerThread(void *arg)
}
}
- pmgmt->processEventQueue();
ret = pmgmt->processSignalQueue();
if (ret < 0 && pmgmt->running) {
Alert("exiting with write error from process manager: %s", strerror(-ret));
}
-
- mgmt_sleep_sec(pmgmt->timeout);
}
return ret;
@@ -183,6 +184,11 @@ ProcessManager::ProcessManager(bool rlm)
{
mgmt_signal_queue = create_queue();
+ local_manager_sockfd = ts::NO_FD;
+#if HAVE_EVENTFD
+ wakeup_fd = ts::NO_FD;
+#endif
+
// Set temp. process/manager timeout. Will be reconfigure later.
// Making the process_manager thread a spinning thread to start traffic server
// as quickly as possible. Will reset this timeout when reconfigure()
@@ -240,33 +246,20 @@ ProcessManager::signalManager(int msg_id, const char *data_raw, int data_len)
memcpy((char *)mh + sizeof(MgmtMessageHdr), data_raw, data_len);
ink_release_assert(enqueue(mgmt_signal_queue, mh));
-}
-
-bool
-ProcessManager::processEventQueue()
-{
- bool ret = false;
-
- while (!queue_is_empty(mgmt_event_queue)) {
- MgmtMessageHdr *mh = (MgmtMessageHdr *)dequeue(mgmt_event_queue);
-
- Debug("pmgmt", "processing event id '%d' payload=%d", mh->msg_id, mh->data_len);
- if (mh->data_len > 0) {
- executeMgmtCallback(mh->msg_id, (char *)mh + sizeof(MgmtMessageHdr), mh->data_len);
- } else {
- executeMgmtCallback(mh->msg_id, nullptr, 0);
- }
- // A shutdown message is a normal exit, so Alert rather than Fatal.
- if (mh->msg_id == MGMT_EVENT_SHUTDOWN) {
- Alert("exiting on shutdown message");
- }
-
- ats_free(mh);
- ret = true;
+#if HAVE_EVENTFD
+ // we don't care about the actual value of wakeup_fd, so just keep adding 1. just need to
+ // wakeup the fd. also, note that wakeup_fd was initalized to non-blocking so we can
+ // directly write to it without any timeout checking.
+ //
+ // don't tigger if MGMT_EVENT_LIBRECORD because they happen all the time
+ // and don't require a quick response. for MGMT_EVENT_LIBRECORD, rely on timeouts so
+ // traffic_server can spend more time doing other things/
+ uint64_t one = 1;
+ if (wakeup_fd != ts::NO_FD && mh->msg_id != MGMT_SIGNAL_LIBRECORDS) {
+ ATS_UNUSED_RETURN(write(wakeup_fd, &one, sizeof(uint64_t))); // trigger to stop polling
}
-
- return ret;
+#endif
}
int
@@ -330,6 +323,13 @@ ProcessManager::initLMConnection()
Fatal("failed to connect management socket '%s': %s", sockpath.c_str(), strerror(errno));
}
+#if HAVE_EVENTFD
+ wakeup_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
+ if (wakeup_fd < 0) {
+ Fatal("unable to create wakeup eventfd. errno: %s", strerror(errno));
+ }
+#endif
+
data_len = sizeof(pid_t);
mh_full = (MgmtMessageHdr *)alloca(sizeof(MgmtMessageHdr) + data_len);
mh_full->msg_id = MGMT_SIGNAL_PID;
@@ -345,26 +345,67 @@ ProcessManager::initLMConnection()
int
ProcessManager::pollLMConnection()
{
- // Avoid getting stuck enqueuing too many requests in a row, limit to MAX_MSGS_IN_A_ROW.
int count;
+ int ready;
+ struct timeval timeout;
+ fd_set fdlist;
+ // Avoid getting stuck enqueuing too many requests in a row, limit to MAX_MSGS_IN_A_ROW.
for (count = 0; running && count < max_msgs_in_a_row; ++count) {
- MgmtMessageHdr *msg;
- int ret = read_management_message(local_manager_sockfd, HRTIME_SECONDS(1), &msg);
- if (ret < 0) {
- return ret;
+ timeout.tv_sec = 1;
+ timeout.tv_usec = 0;
+
+ FD_ZERO(&fdlist);
+
+ if (local_manager_sockfd != ts::NO_FD) {
+ FD_SET(local_manager_sockfd, &fdlist);
}
- // No message, we are done polling. */
- if (msg == nullptr) {
- break;
+#if HAVE_EVENTFD
+ if (wakeup_fd != ts::NO_FD) {
+ FD_SET(wakeup_fd, &fdlist);
}
+#endif
- Debug("pmgmt", "received message ID %d", msg->msg_id);
- handleMgmtMsgFromLM(msg);
- ats_free(msg);
- }
+ // wait for data on socket
+ ready = mgmt_select(FD_SETSIZE, &fdlist, nullptr, nullptr, &timeout);
+
+ switch (ready) {
+ case 0:
+ // Timed out.
+ return 0;
+ case -1:
+ if (mgmt_transient_error()) {
+ continue;
+ }
+ return -errno;
+ }
+
+ if (local_manager_sockfd != ts::NO_FD && FD_ISSET(local_manager_sockfd, &fdlist)) { /* Message from manager */
+ MgmtMessageHdr *msg;
+ int ret = read_management_message(local_manager_sockfd, &msg);
+ if (ret < 0) {
+ return ret;
+ }
+
+ // No message, we are done polling. */
+ if (msg == nullptr) {
+ return 0;
+ }
+
+ Debug("pmgmt", "received message ID %d", msg->msg_id);
+ handleMgmtMsgFromLM(msg);
+ }
+#if HAVE_EVENTFD
+ else if (wakeup_fd != ts::NO_FD && FD_ISSET(wakeup_fd, &fdlist)) { /* if msg, keep polling for more */
+ // read or else fd will always be set.
+ uint64_t ignore;
+ ATS_UNUSED_RETURN(read(wakeup_fd, &ignore, sizeof(uint64_t)));
+ break;
+ }
+#endif
+ }
Debug("pmgmt", "enqueued %d of max %d messages in a row", count, max_msgs_in_a_row);
return 0;
}
@@ -372,23 +413,27 @@ ProcessManager::pollLMConnection()
void
ProcessManager::handleMgmtMsgFromLM(MgmtMessageHdr *mh)
{
+ ink_assert(mh != nullptr);
+
char *data_raw = (char *)mh + sizeof(MgmtMessageHdr);
+ Debug("pmgmt", "processing event id '%d' payload=%d", mh->msg_id, mh->data_len);
switch (mh->msg_id) {
case MGMT_EVENT_SHUTDOWN:
- signalMgmtEntity(MGMT_EVENT_SHUTDOWN);
+ executeMgmtCallback(MGMT_EVENT_SHUTDOWN, nullptr, 0);
+ Alert("exiting on shutdown message");
break;
case MGMT_EVENT_RESTART:
- signalMgmtEntity(MGMT_EVENT_RESTART);
+ executeMgmtCallback(MGMT_EVENT_RESTART, nullptr, 0);
break;
case MGMT_EVENT_DRAIN:
- signalMgmtEntity(MGMT_EVENT_DRAIN, data_raw, mh->data_len);
+ executeMgmtCallback(MGMT_EVENT_DRAIN, data_raw, mh->data_len);
break;
case MGMT_EVENT_CLEAR_STATS:
- signalMgmtEntity(MGMT_EVENT_CLEAR_STATS);
+ executeMgmtCallback(MGMT_EVENT_CLEAR_STATS, nullptr, 0);
break;
case MGMT_EVENT_ROLL_LOG_FILES:
- signalMgmtEntity(MGMT_EVENT_ROLL_LOG_FILES);
+ executeMgmtCallback(MGMT_EVENT_ROLL_LOG_FILES, nullptr, 0);
break;
case MGMT_EVENT_PLUGIN_CONFIG_UPDATE:
if (data_raw != nullptr && data_raw[0] != '\0' && this->cbtable) {
@@ -413,16 +458,18 @@ ProcessManager::handleMgmtMsgFromLM(MgmtMessageHdr *mh)
*/
break;
case MGMT_EVENT_LIBRECORDS:
- signalMgmtEntity(MGMT_EVENT_LIBRECORDS, data_raw, mh->data_len);
+ executeMgmtCallback(MGMT_EVENT_LIBRECORDS, data_raw, mh->data_len);
break;
case MGMT_EVENT_STORAGE_DEVICE_CMD_OFFLINE:
- signalMgmtEntity(MGMT_EVENT_STORAGE_DEVICE_CMD_OFFLINE, data_raw, mh->data_len);
+ executeMgmtCallback(MGMT_EVENT_STORAGE_DEVICE_CMD_OFFLINE, data_raw, mh->data_len);
break;
case MGMT_EVENT_LIFECYCLE_MESSAGE:
- signalMgmtEntity(MGMT_EVENT_LIFECYCLE_MESSAGE, data_raw, mh->data_len);
+ executeMgmtCallback(MGMT_EVENT_LIFECYCLE_MESSAGE, data_raw, mh->data_len);
break;
default:
Warning("received unknown message ID %d\n", mh->msg_id);
break;
}
+
+ ats_free(mh);
}
diff --git a/mgmt/ProcessManager.h b/mgmt/ProcessManager.h
index b622a75..b3b2120 100644
--- a/mgmt/ProcessManager.h
+++ b/mgmt/ProcessManager.h
@@ -43,6 +43,10 @@
#include "ts/ink_apidefs.h"
#include <functional>
+#if HAVE_EVENTFD
+#include <sys/eventfd.h>
+#endif
+
class ConfigUpdateCbTable;
class ProcessManager : public BaseManager
@@ -91,6 +95,9 @@ private:
std::function<void()> init;
int local_manager_sockfd;
+#if HAVE_EVENTFD
+ int wakeup_fd; // external trigger to stop polling
+#endif
ConfigUpdateCbTable *cbtable;
int max_msgs_in_a_row;
--
To stop receiving notification emails like this one, please contact
shinrich@apache.org.