You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafficserver.apache.org by bc...@apache.org on 2018/11/20 22:42:42 UTC
[trafficserver] branch master updated: Use LoopTailHandler in
UDPNetHandler to support signalActivity
This is an automated email from the ASF dual-hosted git repository.
bcall 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 8c7d6f7 Use LoopTailHandler in UDPNetHandler to support signalActivity
8c7d6f7 is described below
commit 8c7d6f75ab4951f8bc50a27f6a409c85884f6a6d
Author: scw00 <sc...@apache.org>
AuthorDate: Tue Nov 20 11:22:58 2018 +0800
Use LoopTailHandler in UDPNetHandler to support signalActivity
---
iocore/net/P_UDPNet.h | 6 ++++-
iocore/net/UnixUDPNet.cc | 65 ++++++++++++++++++++++++++++++++++++++++--------
2 files changed, 60 insertions(+), 11 deletions(-)
diff --git a/iocore/net/P_UDPNet.h b/iocore/net/P_UDPNet.h
index 8d803a0..9e7d869 100644
--- a/iocore/net/P_UDPNet.h
+++ b/iocore/net/P_UDPNet.h
@@ -303,7 +303,7 @@ public:
void initialize_thread_for_udp_net(EThread *thread);
-class UDPNetHandler : public Continuation
+class UDPNetHandler : public Continuation, public EThread::LoopTailHandler
{
public:
// engine for outgoing packets
@@ -319,12 +319,16 @@ public:
Que(UnixUDPConnection, callback_link) udp_callbacks;
Event *trigger_event = nullptr;
+ EThread *thread = nullptr;
ink_hrtime nextCheck;
ink_hrtime lastCheck;
int startNetEvent(int event, Event *data);
int mainNetEvent(int event, Event *data);
+ int waitForActivity(ink_hrtime timeout) override;
+ void signalActivity() override;
+
UDPNetHandler();
};
diff --git a/iocore/net/UnixUDPNet.cc b/iocore/net/UnixUDPNet.cc
index 3f73b1b..529a38a 100644
--- a/iocore/net/UnixUDPNet.cc
+++ b/iocore/net/UnixUDPNet.cc
@@ -66,8 +66,14 @@ initialize_thread_for_udp_net(EThread *thread)
new ((ink_dummy_for_new *)get_UDPPollCont(thread)) PollCont(thread->mutex);
// The UDPNetHandler cannot be accessed across EThreads.
// Because the UDPNetHandler should be called back immediately after UDPPollCont.
- nh->mutex = thread->mutex.get();
-
+ nh->mutex = thread->mutex.get();
+ nh->thread = thread;
+
+ PollCont *upc = get_UDPPollCont(thread);
+ PollDescriptor *upd = upc->pollDescriptor;
+ // due to ET_UDP is really simple, it should sleep for a long time
+ // TODO: fixed size
+ upc->poll_timeout = 100;
// This variable controls how often we cleanup the cancelled packets.
// If it is set to 0, then cleanup never occurs.
REC_ReadConfigInt32(g_udp_periodicFreeCancelledPkts, "proxy.config.udp.free_cancelled_pkts_sec");
@@ -82,8 +88,15 @@ initialize_thread_for_udp_net(EThread *thread)
REC_ReadConfigInt32(g_udp_numSendRetries, "proxy.config.udp.send_retries");
g_udp_numSendRetries = g_udp_numSendRetries < 0 ? 0 : g_udp_numSendRetries;
- thread->schedule_every(get_UDPPollCont(thread), -9);
- thread->schedule_imm(get_UDPNetHandler(thread));
+ thread->set_tail_handler(nh);
+ thread->ep = (EventIO *)ats_malloc(sizeof(EventIO));
+ new (thread->ep) EventIO();
+ thread->ep->type = EVENTIO_ASYNC_SIGNAL;
+#if HAVE_EVENTFD
+ thread->ep->start(upd, thread->evfd, nullptr, EVENTIO_READ);
+#else
+ thread->ep->start(upd, thread->evpipe[0], nullptr, EVENTIO_READ);
+#endif
}
int
@@ -890,6 +903,20 @@ UDPQueue::send(UDPPacket *p)
#undef LINK
+static void
+net_signal_hook_callback(EThread *thread)
+{
+#if HAVE_EVENTFD
+ uint64_t counter;
+ ATS_UNUSED_RETURN(read(thread->evfd, &counter, sizeof(uint64_t)));
+#elif TS_USE_PORT
+/* Nothing to drain or do */
+#else
+ char dummy[1024];
+ ATS_UNUSED_RETURN(read(thread->evpipe[0], &dummy[0], 1024));
+#endif
+}
+
UDPNetHandler::UDPNetHandler()
{
nextCheck = Thread::get_hrtime_updated() + HRTIME_MSECONDS(1000);
@@ -911,10 +938,15 @@ int
UDPNetHandler::mainNetEvent(int event, Event *e)
{
ink_assert(trigger_event == e && event == EVENT_POLL);
- (void)event;
- (void)e;
+ return this->waitForActivity(net_config_poll_timeout);
+}
+int
+UDPNetHandler::waitForActivity(ink_hrtime timeout)
+{
UnixUDPConnection *uc;
+ PollCont *pc = get_UDPPollCont(this->thread);
+ pc->do_poll(timeout);
/* Notice: the race between traversal of newconn_list and UDPBind()
*
@@ -943,7 +975,6 @@ UDPNetHandler::mainNetEvent(int event, Event *e)
// handle UDP read operations
int i, nread = 0;
- PollCont *pc = get_UDPPollCont(e->ethread);
EventIO *epd = nullptr;
for (i = 0; i < pc->pollDescriptor->result; i++) {
epd = (EventIO *)get_ev_data(pc->pollDescriptor, i);
@@ -973,8 +1004,7 @@ UDPNetHandler::mainNetEvent(int event, Event *e)
#endif
}
} else if (epd->type == EVENTIO_ASYNC_SIGNAL) {
- // TODO: receive signal from event system
- // net_signal_hook_callback(this->trigger_event->ethread);
+ net_signal_hook_callback(this->thread);
}
} // end for
@@ -997,7 +1027,7 @@ UDPNetHandler::mainNetEvent(int event, Event *e)
udp_callbacks.clear();
while ((uc = q.dequeue())) {
ink_assert(uc->mutex && uc->continuation);
- if (udpNetInternal.udp_callback(this, uc, trigger_event->ethread)) { // not successful
+ if (udpNetInternal.udp_callback(this, uc, this->thread)) { // not successful
// schedule on a thread of its own.
ink_assert(uc->callback_link.next == nullptr);
ink_assert(uc->callback_link.prev == nullptr);
@@ -1012,3 +1042,18 @@ UDPNetHandler::mainNetEvent(int event, Event *e)
return EVENT_CONT;
}
+
+void
+UDPNetHandler::signalActivity()
+{
+#if HAVE_EVENTFD
+ uint64_t counter = 1;
+ ATS_UNUSED_RETURN(write(thread->evfd, &counter, sizeof(uint64_t)));
+#elif TS_USE_PORT
+ PollDescriptor *pd = get_PollDescriptor(thread);
+ ATS_UNUSED_RETURN(port_send(pd->port_fd, 0, thread->ep));
+#else
+ char dummy = 1;
+ ATS_UNUSED_RETURN(write(thread->evpipe[1], &dummy, 1));
+#endif
+}