Line data Source code
1 : //
2 : // Copyright (c) 2016 Juniper Networks, Inc. All rights reserved.
3 : // Copyright Contributors to the OpenSDN project
4 : //
5 :
6 : #include <cerrno>
7 : #include <cstring>
8 :
9 : #include <base/logging.h>
10 :
11 : #include <zookeeper/zookeeper_client.h>
12 : #include <zookeeper/zookeeper_client_impl.h>
13 : #include <zookeeper/zookeeper_interface.h>
14 :
15 : #define ZOO_LOG(_Level, _Msg) \
16 : do { \
17 : if (LoggingDisabled()) break; \
18 : log4cplus::Logger logger = log4cplus::Logger::getRoot(); \
19 : LOG4CPLUS_##_Level(logger, __func__ << ":" << __FILE__ << ":" << \
20 : __LINE__ << ": " << _Msg); \
21 : } while (false)
22 :
23 : #define ZOO_LOG_ERR(_Msg) \
24 : do { \
25 : LOG(ERROR, __func__ << ":" << __FILE__ << ":" << __LINE__ << ": " \
26 : << _Msg); \
27 : } while (false)
28 :
29 : namespace zookeeper {
30 : namespace interface {
31 :
32 : class ZookeeperCBindings : public ZookeeperInterface {
33 : public:
34 0 : ZookeeperCBindings() {
35 0 : }
36 0 : virtual ~ZookeeperCBindings() {
37 0 : }
38 0 : virtual void ZooSetDebugLevel(ZooLogLevel logLevel) {
39 0 : zoo_set_debug_level(logLevel);
40 0 : }
41 0 : virtual zhandle_t* ZookeeperInit(const char *host, watcher_fn fn,
42 : int recv_timeout, const clientid_t *clientid, void *context,
43 : int flags) {
44 0 : return zookeeper_init(host, fn, recv_timeout, clientid, context,
45 0 : flags);
46 : }
47 0 : virtual int ZookeeperClose(zhandle_t *zh) {
48 0 : return zookeeper_close(zh);
49 : }
50 0 : virtual int ZooState(zhandle_t *zh) {
51 0 : return zoo_state(zh);
52 : }
53 0 : virtual int ZooCreate(zhandle_t *zh, const char *path, const char *value,
54 : int valuelen, const struct ACL_vector *acl, int flags,
55 : char *path_buffer, int path_buffer_len) {
56 0 : return zoo_create(zh, path, value, valuelen, acl, flags, path_buffer,
57 0 : path_buffer_len);
58 : }
59 0 : virtual int ZooDelete(zhandle_t *zh, const char *path, int version) {
60 0 : return zoo_delete(zh, path, version);
61 : }
62 0 : virtual int ZooGet(zhandle_t *zh, const char *path, int watch,
63 : char *buffer, int* buffer_len, struct Stat *stat) {
64 0 : return zoo_get(zh, path, watch, buffer, buffer_len, stat);
65 : }
66 0 : virtual int ZooExists(zhandle_t *zh, const char *path, int watch,
67 : struct Stat *stat) {
68 0 : return zoo_exists(zh, path, watch, stat);
69 : }
70 0 : virtual void ZooSetContext(zhandle_t * zh, void *context) {
71 0 : return zoo_set_context(zh, context);
72 : }
73 0 : virtual int ZooIsUnrecoverable(zhandle_t * zh) {
74 0 : return is_unrecoverable(zh);
75 : }
76 : };
77 :
78 : } // namespace interface
79 :
80 : namespace client {
81 : namespace impl {
82 :
83 0 : void ZookeeperWatcher(zhandle_t* zh, int type, int state,
84 : const char* path, void* watcherCtx) {
85 0 : if (ZINVALIDSTATE == is_unrecoverable(zh)) {
86 0 : ZOO_LOG(DEBUG, "Zookeeper callback called with state " << state);
87 0 : ZookeeperClientImpl *zooImpl = (ZookeeperClientImpl *)watcherCtx;
88 0 : if (zooImpl && zooImpl->GetClient()) {
89 0 : if (((ZookeeperClient *)zooImpl->GetClient())->cb) {
90 0 : ((ZookeeperClient *)zooImpl->GetClient())->cb();
91 : }
92 : }
93 : }
94 0 : }
95 :
96 : // ZookeeperClientImpl
97 6 : ZookeeperClientImpl::ZookeeperClientImpl(const char *hostname,
98 6 : const char *servers, zookeeper::interface::ZookeeperInterface *zki) :
99 6 : hostname_(hostname),
100 6 : servers_(servers),
101 6 : zk_handle_(NULL),
102 6 : connected_(false),
103 12 : zki_(zki) {
104 : // Set loglevel
105 6 : zki_->ZooSetDebugLevel(ZOO_LOG_LEVEL_DEBUG);
106 6 : }
107 :
108 12 : ZookeeperClientImpl::~ZookeeperClientImpl() {
109 12 : }
110 :
111 7 : bool ZookeeperClientImpl::Connect() {
112 : while (true) {
113 7 : zk_handle_ = zki_->ZookeeperInit(servers_.c_str(),
114 : ZookeeperWatcher,
115 : kSessionTimeoutMSec_,
116 : NULL,
117 : NULL,
118 : 0);
119 7 : if (zk_handle_ == NULL) {
120 0 : int zerrno(errno);
121 0 : ZOO_LOG_ERR("zookeeper_init FAILED: (" << zerrno <<
122 : "): servers: " << servers_ << " retrying in 1 second");
123 0 : sleep(1);
124 0 : continue;
125 0 : }
126 7 : zki_->ZooSetContext(zk_handle_, this);
127 : // Block till session is connected
128 8 : while (!connected_) {
129 8 : int zstate(zki_->ZooState(zk_handle_));
130 8 : if (zstate == ZOO_CONNECTED_STATE) {
131 7 : connected_ = true;
132 7 : ZOO_LOG(DEBUG, "Session CONNECTED");
133 7 : break;
134 : } else {
135 1 : ZOO_LOG(DEBUG, "Session NOT CONNECTED: retrying in 1 second");
136 1 : sleep(1);
137 1 : continue;
138 1 : }
139 : }
140 7 : break;
141 0 : }
142 7 : assert(connected_);
143 7 : return true;
144 : }
145 :
146 7 : void ZookeeperClientImpl::Shutdown() {
147 7 : if (zk_handle_) {
148 7 : int rc(zki_->ZookeeperClose(zk_handle_));
149 7 : if (rc != ZOK) {
150 0 : int zerrno(errno);
151 0 : ZOO_LOG(WARN, "zookeeper_close FAILED (" << rc <<
152 : "): errno: " << zerrno);
153 : }
154 7 : zk_handle_ = NULL;
155 : }
156 7 : connected_ = false;
157 7 : }
158 :
159 1 : bool ZookeeperClientImpl::Reconnect() {
160 1 : Shutdown();
161 1 : return Connect();
162 : }
163 :
164 19 : bool ZookeeperClientImpl::IsConnected() const {
165 19 : return connected_;
166 : }
167 :
168 20 : static inline bool IsZooErrorRecoverable(int zerror) {
169 20 : return zerror == ZCONNECTIONLOSS ||
170 20 : zerror == ZOPERATIONTIMEOUT;
171 : }
172 :
173 16 : static inline bool IsZooErrorUnrecoverable(int zerror) {
174 16 : return zerror == ZINVALIDSTATE;
175 : }
176 :
177 8 : int ZookeeperClientImpl::CreateNodeSync(const char *path, const char *value,
178 : int *err, int flag) {
179 : int rc;
180 8 : retry:
181 : do {
182 10 : rc = zki_->ZooCreate(zk_handle_, path, value, strlen(value),
183 : &ZOO_OPEN_ACL_UNSAFE, flag, NULL, -1);
184 10 : } while (IsZooErrorRecoverable(rc));
185 8 : if (IsZooErrorUnrecoverable(rc)) {
186 : // Reconnect
187 1 : Reconnect();
188 1 : goto retry;
189 : }
190 7 : if (rc != ZOK) {
191 2 : *err = errno;
192 : }
193 7 : return rc;
194 : }
195 :
196 0 : bool ZookeeperClientImpl::CreateNode(const char *path, const char *value,
197 : int flag) {
198 0 : int err = 0;
199 : int rc;
200 0 : if (!IsConnected()) {
201 0 : bool success(Connect());
202 0 : if (!success) {
203 0 : ZOO_LOG_ERR("Zookeeper Client Connect FAILED");
204 0 : return false;
205 : }
206 : }
207 0 : rc = CreateNodeSync(path, value, &err, flag);
208 0 : if (rc != ZOK && rc != ZNODEEXISTS) {
209 0 : ZOO_LOG_ERR("Creation of ZNODE(" << path << "): " << value
210 : << ": FAILED: (" << rc << ") error: " << err);
211 0 : return false;
212 : }
213 0 : return true;
214 : }
215 :
216 2 : int ZookeeperClientImpl::GetNodeDataSync(const char *path, char *buf,
217 : int *buf_len, int *err) {
218 : int rc;
219 2 : retry:
220 : do {
221 2 : rc = zki_->ZooGet(zk_handle_, path, 0, buf, buf_len, NULL);
222 2 : } while (IsZooErrorRecoverable(rc));
223 2 : if (IsZooErrorUnrecoverable(rc)) {
224 : // Reconnect
225 0 : Reconnect();
226 0 : goto retry;
227 : }
228 2 : if (rc != ZOK) {
229 0 : *err = errno;
230 : }
231 2 : return rc;
232 : }
233 :
234 0 : bool ZookeeperClientImpl::CheckNodeExist(const char *path) {
235 0 : if (!IsConnected()) {
236 0 : bool success(Connect());
237 0 : if (!success) {
238 0 : ZOO_LOG_ERR("Zookeeper Client Connect FAILED");
239 0 : return false;
240 : }
241 : }
242 :
243 : struct Stat stat;
244 0 : int rc = zki_->ZooExists(zk_handle_, path, 0, &stat);
245 0 : return (rc == ZOK);
246 : }
247 :
248 6 : int ZookeeperClientImpl::DeleteNodeSync(const char *path, int *err) {
249 : int rc;
250 6 : retry:
251 : do {
252 8 : rc = zki_->ZooDelete(zk_handle_, path, -1);
253 8 : } while (IsZooErrorRecoverable(rc));
254 6 : if (IsZooErrorUnrecoverable(rc)) {
255 : // Reconnect
256 0 : Reconnect();
257 0 : goto retry;
258 : }
259 6 : if (rc != ZOK) {
260 0 : *err = errno;
261 : }
262 6 : return rc;
263 : }
264 :
265 0 : bool ZookeeperClientImpl::DeleteNode(const char *path) {
266 0 : int err = 0;
267 : int rc;
268 0 : if (!IsConnected()) {
269 0 : bool success(Connect());
270 0 : if (!success) {
271 0 : ZOO_LOG_ERR("Zookeeper Client Connect FAILED");
272 0 : return false;
273 : }
274 : }
275 :
276 0 : rc = DeleteNodeSync(path, &err);
277 0 : if (rc != ZOK) {
278 0 : ZOO_LOG_ERR("Deletion of ZNODE(" << path << "): "
279 : << ": FAILED: (" << rc << ") error: " << err);
280 0 : return false;
281 : }
282 :
283 0 : return (rc == ZOK);
284 : }
285 :
286 6 : std::string ZookeeperClientImpl::Name() const {
287 6 : return hostname_;
288 : }
289 :
290 : } // namespace impl
291 :
292 : // ZookeeperClient
293 0 : ZookeeperClient::ZookeeperClient(const char *hostname, const char *servers) :
294 0 : impl_(new impl::ZookeeperClientImpl(hostname, servers,
295 0 : new zookeeper::interface::ZookeeperCBindings)) {
296 0 : }
297 :
298 6 : ZookeeperClient::ZookeeperClient(impl::ZookeeperClientImpl *impl) :
299 6 : impl_(impl) {
300 6 : }
301 :
302 0 : bool ZookeeperClient::CreateNode(const char *path, const char *value,
303 : int type) {
304 0 : int flag = 0;
305 0 : if (type == Z_NODE_TYPE_EPHEMERAL) {
306 0 : flag |= ZOO_EPHEMERAL;
307 : }
308 0 : if (type == Z_NODE_TYPE_SEQUENCE) {
309 0 : flag |= ZOO_SEQUENCE;
310 : }
311 0 : return impl_->CreateNode(path, value, flag);
312 : }
313 :
314 0 : bool ZookeeperClient::CheckNodeExist(const char *path) {
315 0 : return impl_->CheckNodeExist(path);
316 : }
317 :
318 0 : bool ZookeeperClient::DeleteNode(const char *path) {
319 0 : return impl_->DeleteNode(path);
320 : }
321 :
322 0 : void ZookeeperClient::Shutdown() {
323 0 : return impl_->Shutdown();
324 : }
325 :
326 0 : void ZookeeperClient::AddListener(ZooStateCallback callback) {
327 0 : cb = callback;
328 0 : if (impl_.get()) {
329 0 : impl_->SetClient(this);
330 : }
331 0 : }
332 :
333 12 : ZookeeperClient::~ZookeeperClient() {
334 12 : }
335 :
336 : // ZookeeperLockImpl
337 : class ZookeeperLock::ZookeeperLockImpl {
338 : public:
339 6 : ZookeeperLockImpl(impl::ZookeeperClientImpl *clientImpl,
340 6 : const char *path) :
341 6 : clientImpl_(clientImpl),
342 6 : path_(path),
343 6 : is_acquired_(false) {
344 6 : id_ = clientImpl_->Name();
345 6 : }
346 :
347 6 : std::string Id() const {
348 6 : return id_;
349 : }
350 :
351 6 : bool Lock() {
352 6 : ZOO_LOG(INFO, "Trying (" << path_ << "): " << id_);
353 : while (true) {
354 : // Connect if not already done
355 7 : if (!clientImpl_->IsConnected()) {
356 6 : bool success(clientImpl_->Connect());
357 6 : if (!success) {
358 0 : ZOO_LOG_ERR("Zookeeper Client Connect FAILED");
359 6 : return success;
360 : }
361 : }
362 : // Try creating the znode
363 : int err;
364 7 : int rc(clientImpl_->CreateNodeSync(path_.c_str(), id_.c_str(),
365 : &err, 0));
366 7 : switch (rc) {
367 5 : case ZOK: {
368 : // We acquired the lock
369 5 : ZOO_LOG(INFO, "ACQUIRED (" << path_ << "): " << id_);
370 5 : is_acquired_ = true;
371 5 : return true;
372 : }
373 2 : case ZNODEEXISTS: {
374 : // Node exists, get node data and check
375 : char buf[256];
376 2 : int buf_len(sizeof(buf));
377 : int zerr;
378 2 : int zrc(clientImpl_->GetNodeDataSync(path_.c_str(), buf,
379 : &buf_len, &zerr));
380 2 : if (zrc == ZOK) {
381 : // Does it match our ID?
382 2 : std::string mid(buf, buf_len);
383 2 : if (id_ == mid) {
384 : // We acquired the lock
385 1 : ZOO_LOG(INFO, "ACQUIRED EEXIST (" << path_ << "): "
386 : << id_);
387 1 : is_acquired_ = true;
388 1 : return true;
389 : }
390 1 : ZOO_LOG_ERR("EEXIST (" << path_ << "): " << mid <<
391 : " , ours: " << id_);
392 1 : sleep(1);
393 1 : continue;
394 2 : } else if (zrc == ZNONODE) {
395 0 : ZOO_LOG(WARN, "GetNodeDataSync(" << path_ <<
396 : "): Data: " << id_ <<
397 : ": No Node EXISTS: retrying in 1 second");
398 0 : sleep(1);
399 0 : continue;
400 0 : } else {
401 0 : ZOO_LOG_ERR("GetNodeDataSync(" << path_ << "): " <<
402 : id_ << ": FAILED: (" << zrc << ") error: " << zerr);
403 0 : clientImpl_->Shutdown();
404 0 : return false;
405 : }
406 : break;
407 : }
408 0 : default: {
409 0 : ZOO_LOG_ERR("CreateNodeSync(" << path_ << "): " << id_
410 : << ": FAILED: (" << rc << ") error: " << err);
411 0 : clientImpl_->Shutdown();
412 0 : return false;
413 : }
414 : }
415 1 : }
416 : }
417 :
418 6 : bool Release() {
419 : bool success;
420 : int err, rc;
421 6 : if (!is_acquired_) {
422 0 : ZOO_LOG_ERR("(" << path_ << "): " << id_ <<
423 : ": Release WITHOUT Lock");
424 0 : success = false;
425 0 : goto cleanup;
426 : }
427 6 : is_acquired_ = false;
428 6 : rc = clientImpl_->DeleteNodeSync(path_.c_str(), &err);
429 6 : if (rc == ZOK) {
430 6 : ZOO_LOG(INFO, "RELEASED (" << path_ << "): " << id_);
431 6 : success = true;
432 6 : goto cleanup;
433 0 : } else if (rc == ZNONODE) {
434 0 : ZOO_LOG_ERR("DeleteNodeSync(" << path_ << "): " << id_ <<
435 : ": No Node EXISTS(" << err <<
436 : "): Possible concurrent execution");
437 0 : success = false;
438 0 : goto cleanup;
439 : } else {
440 0 : ZOO_LOG_ERR("DeleteNodeSync(" << path_ << "): " << id_ <<
441 : ": FAILED (" << rc << "): error " << err);
442 0 : success = false;
443 0 : goto cleanup;
444 : }
445 6 : cleanup:
446 6 : clientImpl_->Shutdown();
447 6 : return success;
448 : }
449 :
450 : private:
451 : impl::ZookeeperClientImpl *clientImpl_;
452 : std::string path_;
453 : bool is_acquired_;
454 : std::string id_;
455 : };
456 :
457 : // ZookeeperLock
458 6 : ZookeeperLock::ZookeeperLock(ZookeeperClient *client, const char *path) :
459 6 : impl_(new ZookeeperLockImpl(client->impl_.get(), path)) {
460 6 : }
461 :
462 6 : ZookeeperLock::~ZookeeperLock() {
463 6 : }
464 :
465 6 : std::string ZookeeperLock::Id() const {
466 6 : return impl_->Id();
467 : }
468 :
469 6 : bool ZookeeperLock::Lock() {
470 6 : return impl_->Lock();
471 : }
472 :
473 6 : bool ZookeeperLock::Release() {
474 6 : return impl_->Release();
475 : }
476 :
477 : } // namespace client
478 : } // namespace zookeeper
|
