"Fossies" - the Fresh Open Source Software Archive

Member "apache-zookeeper-3.5.6/zookeeper-client/zookeeper-client-c/src/zk_adaptor.h" (5 Oct 2019, 10921 Bytes) of package /linux/misc/apache-zookeeper-3.5.6.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) C and C++ source code syntax highlighting (style: standard) with prefixed line numbers and code folding option. Alternatively you can here view or download the uninterpreted source code file. For more information about "zk_adaptor.h" see the Fossies "Dox" file reference documentation.

    1 /**
    2  * Licensed to the Apache Software Foundation (ASF) under one
    3  * or more contributor license agreements.  See the NOTICE file
    4  * distributed with this work for additional information
    5  * regarding copyright ownership.  The ASF licenses this file
    6  * to you under the Apache License, Version 2.0 (the
    7  * "License"); you may not use this file except in compliance
    8  * with the License.  You may obtain a copy of the License at
    9  *
   10  *     http://www.apache.org/licenses/LICENSE-2.0
   11  *
   12  * Unless required by applicable law or agreed to in writing, software
   13  * distributed under the License is distributed on an "AS IS" BASIS,
   14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   15  * See the License for the specific language governing permissions and
   16  * limitations under the License.
   17  */
   18 
   19 #ifndef ZK_ADAPTOR_H_
   20 #define ZK_ADAPTOR_H_
   21 #include <zookeeper.jute.h>
   22 #ifdef THREADED
   23 #ifndef WIN32
   24 #include <pthread.h>
   25 #else
   26 #include "winport.h"
   27 #endif
   28 #endif
   29 #include "zookeeper.h"
   30 #include "zk_hashtable.h"
   31 #include "addrvec.h"
   32 
   33 /* predefined xid's values recognized as special by the server */
   34 #define WATCHER_EVENT_XID -1 
   35 #define PING_XID -2
   36 #define AUTH_XID -4
   37 #define SET_WATCHES_XID -8
   38 
   39 /* zookeeper state constants */
   40 #define EXPIRED_SESSION_STATE_DEF -112
   41 #define AUTH_FAILED_STATE_DEF -113
   42 #define CONNECTING_STATE_DEF 1
   43 #define ASSOCIATING_STATE_DEF 2
   44 #define CONNECTED_STATE_DEF 3
   45 #define READONLY_STATE_DEF 5
   46 #define NOTCONNECTED_STATE_DEF 999
   47 
   48 /* zookeeper event type constants */
   49 #define CREATED_EVENT_DEF 1
   50 #define DELETED_EVENT_DEF 2
   51 #define CHANGED_EVENT_DEF 3
   52 #define CHILD_EVENT_DEF 4
   53 #define SESSION_EVENT_DEF -1
   54 #define NOTWATCHING_EVENT_DEF -2
   55 
   56 #ifdef __cplusplus
   57 extern "C" {
   58 #endif
   59 
   60 struct _buffer_list;
   61 struct _completion_list;
   62 
   63 typedef struct _buffer_head {
   64     struct _buffer_list *volatile head;
   65     struct _buffer_list *last;
   66 #ifdef THREADED
   67     pthread_mutex_t lock;
   68 #endif
   69 } buffer_head_t;
   70 
   71 typedef struct _completion_head {
   72     struct _completion_list *volatile head;
   73     struct _completion_list *last;
   74 #ifdef THREADED
   75     pthread_cond_t cond;
   76     pthread_mutex_t lock;
   77 #endif
   78 } completion_head_t;
   79 
   80 int lock_buffer_list(buffer_head_t *l);
   81 int unlock_buffer_list(buffer_head_t *l);
   82 int lock_completion_list(completion_head_t *l);
   83 int unlock_completion_list(completion_head_t *l);
   84 
   85 struct sync_completion {
   86     int rc;
   87     union {
   88         struct {
   89             char *str;
   90             int str_len;
   91         } str;
   92         struct Stat stat;
   93         struct {
   94             char *buffer;
   95             int buff_len;
   96             struct Stat stat;
   97         } data;
   98         struct {
   99             struct ACL_vector acl;
  100             struct Stat stat;
  101         } acl;
  102         struct String_vector strs2;
  103         struct {
  104             struct String_vector strs2;
  105             struct Stat stat2;
  106         } strs_stat;
  107     } u;
  108     int complete;
  109 #ifdef THREADED
  110     pthread_cond_t cond;
  111     pthread_mutex_t lock;
  112 #endif
  113 };
  114 
  115 typedef struct _auth_info {
  116     int state; /* 0=>inactive, >0 => active */
  117     char* scheme;
  118     struct buffer auth;
  119     void_completion_t completion;
  120     const char* data;
  121     struct _auth_info *next;
  122 } auth_info;
  123 
  124 /**
  125  * This structure represents a packet being read or written.
  126  */
  127 typedef struct _buffer_list {
  128     char *buffer;
  129     int len; /* This represents the length of sizeof(header) + length of buffer */
  130     int curr_offset; /* This is the offset into the header followed by offset into the buffer */
  131     struct _buffer_list *next;
  132 } buffer_list_t;
  133 
  134 /* the size of connect request */
  135 #define HANDSHAKE_REQ_SIZE 45
  136 /* connect request */
  137 struct connect_req {
  138     int32_t protocolVersion;
  139     int64_t lastZxidSeen;
  140     int32_t timeOut;
  141     int64_t sessionId;
  142     int32_t passwd_len;
  143     char passwd[16];
  144     char readOnly;
  145 };
  146 
  147 /* the connect response */
  148 struct prime_struct {
  149     int32_t len;
  150     int32_t protocolVersion;
  151     int32_t timeOut;
  152     int64_t sessionId;
  153     int32_t passwd_len;
  154     char passwd[16];
  155     char readOnly;
  156 }; 
  157 
  158 #ifdef THREADED
  159 /* this is used by mt_adaptor internally for thread management */
  160 struct adaptor_threads {
  161      pthread_t io;
  162      pthread_t completion;
  163      int threadsToWait;             // barrier
  164      pthread_cond_t cond;           // barrier's conditional
  165      pthread_mutex_t lock;          // ... and a lock
  166      pthread_mutex_t zh_lock;       // critical section lock
  167      pthread_mutex_t reconfig_lock; // lock for reconfiguring cluster's ensemble
  168 #ifdef WIN32
  169      SOCKET self_pipe[2];
  170 #else
  171      int self_pipe[2];
  172 #endif
  173 };
  174 #endif
  175 
  176 /** the auth list for adding auth */
  177 typedef struct _auth_list_head {
  178      auth_info *auth;
  179 #ifdef THREADED
  180      pthread_mutex_t lock;
  181 #endif
  182 } auth_list_head_t;
  183 
  184 /**
  185  * This structure represents the connection to zookeeper.
  186  */
  187 struct _zhandle {
  188 #ifdef WIN32
  189     SOCKET fd;                          // the descriptor used to talk to zookeeper
  190 #else
  191     int fd;                             // the descriptor used to talk to zookeeper
  192 #endif
  193 
  194     // Hostlist and list of addresses
  195     char *hostname;                     // hostname contains list of zookeeper servers to connect to
  196     struct sockaddr_storage addr_cur;   // address of server we're currently connecting/connected to 
  197     struct sockaddr_storage addr_rw_server; // address of last known read/write server found.
  198 
  199     addrvec_t addrs;                    // current list of addresses we're connected to
  200     addrvec_t addrs_old;                // old list of addresses that we are no longer connected to
  201     addrvec_t addrs_new;                // new list of addresses to connect to if we're reconfiguring
  202 
  203     int reconfig;                       // Are we in the process of reconfiguring cluster's ensemble
  204     double pOld, pNew;                  // Probability for selecting between 'addrs_old' and 'addrs_new'
  205     int delay;
  206     int disable_reconnection_attempt;   // When set, client will not try reconnect to a different server in
  207                                         // server list. This makes a sticky server for client, and is useful
  208                                         // for testing if a sticky server is required, or if client wants to
  209                                         // explicitly shuffle server by calling zoo_cycle_next_server.
  210                                         // The default value is 0.
  211 
  212     watcher_fn watcher;                 // the registered watcher
  213 
  214     // Message timings
  215     struct timeval last_recv;           // time last message was received
  216     struct timeval last_send;           // time last message was sent
  217     struct timeval last_ping;           // time last PING was sent
  218     struct timeval next_deadline;       // time of the next deadline
  219     int recv_timeout;                   // max receive timeout for messages from server
  220 
  221     // Buffers
  222     buffer_list_t *input_buffer;        // current buffer being read in
  223     buffer_head_t to_process;           // buffers that have been read and ready to be processed
  224     buffer_head_t to_send;              // packets queued to send
  225     completion_head_t sent_requests;    // outstanding requests
  226     completion_head_t completions_to_process; // completions that are ready to run
  227     int outstanding_sync;               // number of outstanding synchronous requests
  228 
  229     /* read-only mode specific fields */
  230     struct timeval last_ping_rw; /* The last time we checked server for being r/w */
  231     int ping_rw_timeout; /* The time that can go by before checking next server */
  232 
  233     // State info
  234     volatile int state;                 // Current zookeeper state
  235     void *context;                      // client-side provided context
  236     clientid_t client_id;               // client-id
  237     long long last_zxid;                // last zookeeper ID
  238     auth_list_head_t auth_h;            // authentication data list
  239     log_callback_fn log_callback;       // Callback for logging (falls back to logging to stderr)
  240     int io_count;           // counts the number of iterations of do_io
  241 
  242     // Primer storage
  243     struct _buffer_list primer_buffer;  // The buffer used for the handshake at the start of a connection
  244     struct prime_struct primer_storage; // the connect response
  245     char primer_storage_buffer[41];     // the true size of primer_storage
  246 
  247     /* zookeeper_close is not reentrant because it de-allocates the zhandler. 
  248      * This guard variable is used to defer the destruction of zhandle till 
  249      * right before top-level API call returns to the caller */
  250     int32_t ref_counter;
  251     volatile int close_requested;
  252     void *adaptor_priv;
  253 
  254     /* Used for debugging only: non-zero value indicates the time when the zookeeper_process
  255      * call returned while there was at least one unprocessed server response 
  256      * available in the socket recv buffer */
  257     struct timeval socket_readable;
  258 
  259     // Watchers
  260     zk_hashtable* active_node_watchers;   
  261     zk_hashtable* active_exist_watchers;
  262     zk_hashtable* active_child_watchers;
  263 
  264     /** used for chroot path at the client side **/
  265     char *chroot;
  266 
  267     /** Indicates if this client is allowed to go to r/o mode */
  268     char allow_read_only;
  269     /** Indicates if we connected to a majority server before */
  270     char seen_rw_server_before;
  271 };
  272 
  273 
  274 int adaptor_init(zhandle_t *zh);
  275 void adaptor_finish(zhandle_t *zh);
  276 void adaptor_destroy(zhandle_t *zh);
  277 #if THREADED
  278 struct sync_completion *alloc_sync_completion(void);
  279 int wait_sync_completion(struct sync_completion *sc);
  280 void free_sync_completion(struct sync_completion *sc);
  281 void notify_sync_completion(struct sync_completion *sc);
  282 #endif
  283 int adaptor_send_queue(zhandle_t *zh, int timeout);
  284 int process_async(int outstanding_sync);
  285 void process_completions(zhandle_t *zh);
  286 int flush_send_queue(zhandle_t*zh, int timeout);
  287 char* sub_string(zhandle_t *zh, const char* server_path);
  288 void free_duplicate_path(const char* free_path, const char* path);
  289 int zoo_lock_auth(zhandle_t *zh);
  290 int zoo_unlock_auth(zhandle_t *zh);
  291 
  292 // ensemble reconfigure access guards
  293 int lock_reconfig(struct _zhandle *zh);
  294 int unlock_reconfig(struct _zhandle *zh);
  295 
  296 // critical section guards
  297 int enter_critical(zhandle_t* zh);
  298 int leave_critical(zhandle_t* zh);
  299 
  300 // zhandle object reference counting
  301 void api_prolog(zhandle_t* zh);
  302 int api_epilog(zhandle_t *zh, int rc);
  303 int32_t get_xid();
  304 
  305 // returns the new value of the ref counter
  306 int32_t inc_ref_counter(zhandle_t* zh,int i);
  307 
  308 #ifdef THREADED
  309 // atomic post-increment
  310 int32_t fetch_and_add(volatile int32_t* operand, int incr);
  311 // in mt mode process session event asynchronously by the completion thread
  312 #define PROCESS_SESSION_EVENT(zh,newstate) queue_session_event(zh,newstate)
  313 #else
  314 // in single-threaded mode process session event immediately
  315 //#define PROCESS_SESSION_EVENT(zh,newstate) deliverWatchers(zh,ZOO_SESSION_EVENT,newstate,0)
  316 #define PROCESS_SESSION_EVENT(zh,newstate) queue_session_event(zh,newstate)
  317 #endif
  318 
  319 #ifdef __cplusplus
  320 }
  321 #endif
  322 
  323 #endif /*ZK_ADAPTOR_H_*/
  324 
  325