Line data Source code
1 : /*
2 : * Copyright (c) 2015 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #ifndef _GNU_SOURCE
7 : #define _GNU_SOURCE
8 : #endif
9 : #include <sched.h>
10 :
11 : #include <unistd.h>
12 : #include <stdlib.h>
13 : #include <sys/eventfd.h>
14 :
15 : #include <algorithm>
16 : #include <vector>
17 : #include <set>
18 : #include <boost/algorithm/string/case_conv.hpp>
19 : #include <boost/thread.hpp>
20 :
21 : #include <tbb/concurrent_queue.h>
22 :
23 : #include "ksync_object.h"
24 : #include "ksync_sock.h"
25 :
26 : static bool ksync_tx_queue_task_done_ = false;
27 :
28 : // Set CPU affinity for KSync Tx Thread based on cpu_pin_policy.
29 : // By default CPU affinity is not set. cpu_pin_policy can change it,
30 : // "last" : Last CPU-ID
31 : // "<num>" : Specifies CPU-ID to pin
32 0 : static void set_thread_affinity(std::string cpu_pin_policy) {
33 0 : int num_cores = boost::thread::hardware_concurrency();
34 0 : if (!num_cores) {
35 0 : LOG(ERROR, "Failure in checking number of available threads");
36 0 : num_cores = 1;
37 : }
38 0 : char *p = NULL;
39 0 : int cpu_id = strtoul(cpu_pin_policy.c_str(), &p, 0);
40 0 : if (*p || cpu_pin_policy.empty()) {
41 : // cpu_pin_policy is non-integer
42 : // Assume pinning disabled by default
43 0 : cpu_id = -1;
44 : // If policy is "last", pick last CPU-ID
45 0 : boost::algorithm::to_lower(cpu_pin_policy);
46 0 : if (cpu_pin_policy == "last") {
47 0 : cpu_id = num_cores - 1;
48 : }
49 : } else {
50 : // cpu_pin_policy is integer
51 : // Disable pinning if configured value out of range
52 0 : if (cpu_id >= num_cores)
53 0 : cpu_id = -1;
54 : }
55 :
56 0 : if (cpu_id >= 0) {
57 : cpu_set_t cpuset;
58 0 : CPU_ZERO(&cpuset);
59 0 : CPU_SET(cpu_id, &cpuset);
60 0 : LOG(ERROR, "KsyncTxQueue CPU pinning policy <" << cpu_pin_policy
61 : << ">. KsyncTxQueue pinned to CPU " << cpu_id);
62 0 : sched_setaffinity(0, sizeof(cpuset), &cpuset);
63 : } else {
64 0 : LOG(ERROR, "KsyncTxQueue CPU pinning policy <" << cpu_pin_policy
65 : << ">. KsyncTxQueuen not pinned to CPU");
66 : }
67 0 : }
68 :
69 : class KSyncTxQueueTask : public Task {
70 : public:
71 0 : KSyncTxQueueTask(TaskScheduler *scheduler, KSyncTxQueue *queue) :
72 0 : Task(scheduler->GetTaskId("Ksync::KSyncTxQueue"), 0), queue_(queue) {
73 0 : }
74 0 : ~KSyncTxQueueTask() {
75 0 : ksync_tx_queue_task_done_ = true;
76 0 : }
77 :
78 0 : bool Run() {
79 0 : queue_->Run();
80 0 : return true;
81 : }
82 0 : std::string Description() const { return "KSyncTxQueue"; }
83 :
84 : private:
85 : KSyncTxQueue *queue_;
86 : };
87 :
88 2 : KSyncTxQueue::KSyncTxQueue(KSyncSock *sock) :
89 2 : work_queue_(NULL),
90 2 : event_fd_(-1),
91 2 : cpu_pin_policy_(),
92 2 : sock_(sock),
93 2 : enqueues_(0),
94 2 : dequeues_(0),
95 2 : write_events_(0),
96 2 : read_events_(0),
97 2 : busy_time_(0),
98 2 : measure_busy_time_(false) {
99 2 : queue_len_ = 0;
100 2 : shutdown_ = false;
101 2 : ClearStats();
102 2 : }
103 :
104 0 : KSyncTxQueue::~KSyncTxQueue() {
105 0 : }
106 :
107 2 : void KSyncTxQueue::Init(bool use_work_queue,
108 : const std::string &cpu_pin_policy) {
109 2 : cpu_pin_policy_ = cpu_pin_policy;
110 2 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
111 2 : if (use_work_queue) {
112 2 : assert(work_queue_ == NULL);
113 4 : work_queue_ = new WorkQueue<IoContext *>
114 4 : (scheduler->GetTaskId("Ksync::AsyncSend"), 0,
115 2 : boost::bind(&KSyncSock::SendAsyncImpl, sock_, _1));
116 2 : work_queue_->SetExitCallback
117 2 : (boost::bind(&KSyncSock::OnEmptyQueue, sock_, _1));
118 2 : return;
119 : }
120 0 : assert((event_fd_ = eventfd(0, (EFD_CLOEXEC | EFD_SEMAPHORE))) >= 0);
121 :
122 0 : KSyncTxQueueTask *task = new KSyncTxQueueTask(scheduler, this);
123 0 : scheduler->Enqueue(task);
124 : }
125 :
126 2 : void KSyncTxQueue::Shutdown() {
127 2 : shutdown_ = true;
128 2 : if (work_queue_) {
129 2 : assert(work_queue_->Length() == 0);
130 2 : work_queue_->Shutdown();
131 2 : delete work_queue_;
132 2 : work_queue_ = NULL;
133 2 : return;
134 : }
135 :
136 0 : uint64_t u = 1;
137 0 : assert(write(event_fd_, &u, sizeof(u)) == sizeof(u));
138 0 : while (queue_len_ != 0) {
139 0 : usleep(1);
140 : }
141 :
142 0 : while(ksync_tx_queue_task_done_ != true) {
143 0 : usleep(1);
144 : }
145 0 : close(event_fd_);
146 : }
147 :
148 11249 : bool KSyncTxQueue::EnqueueInternal(IoContext *io_context) {
149 11249 : if (work_queue_) {
150 11249 : work_queue_->Enqueue(io_context);
151 11249 : return true;
152 : }
153 0 : queue_.push(io_context);
154 0 : enqueues_++;
155 0 : size_t ncount = queue_len_.fetch_add(1) + 1;
156 0 : if (ncount > max_queue_len_)
157 0 : max_queue_len_ = ncount;
158 0 : if (ncount == 1) {
159 0 : uint64_t u = 1;
160 0 : int res = 0;
161 0 : while ((res = write(event_fd_, &u, sizeof(u))) < (int)sizeof(u)) {
162 0 : int ec = errno;
163 0 : if (ec != EINTR && ec != EIO) {
164 0 : LOG(ERROR, "KsyncTxQueue write failure : " << ec << " : "
165 : << strerror(ec));
166 0 : assert(0);
167 : }
168 : }
169 :
170 0 : write_events_++;
171 : }
172 0 : return true;
173 : }
174 :
175 0 : bool KSyncTxQueue::Run() {
176 0 : set_thread_affinity(cpu_pin_policy_);
177 : while (1) {
178 : while (1) {
179 0 : uint64_t u = 0;
180 0 : ssize_t num = read(event_fd_, &u, sizeof(u));
181 0 : if (num >= (int)sizeof(u)) {
182 0 : break;
183 : }
184 0 : if (errno != EINTR && errno != EIO) {
185 0 : LOG(ERROR, "KsyncTxQueue read failure : " << errno << " : "
186 : << strerror(errno));
187 0 : assert(0);
188 : }
189 0 : }
190 0 : read_events_++;
191 :
192 0 : uint64_t t1 = 0;
193 0 : if (measure_busy_time_)
194 0 : t1 = ClockMonotonicUsec();
195 0 : IoContext *io_context = NULL;
196 0 : while (queue_.try_pop(io_context)) {
197 0 : dequeues_++;
198 0 : queue_len_ -= 1;
199 0 : sock_->SendAsyncImpl(io_context);
200 : }
201 0 : sock_->OnEmptyQueue(false);
202 0 : if (shutdown_) {
203 0 : break;
204 : }
205 :
206 0 : if (t1)
207 0 : busy_time_ += (ClockMonotonicUsec() - t1);
208 0 : }
209 0 : return true;
210 : }
|
