首页 > 开发 > 综合 > 正文

aerospike异步处理

2024-07-21 02:52:48
字体:
来源:转载
供稿:网友

1、异步

1、集群在启动时会初始化出异步线程连接队列,其中cluster->async_max_conns_per_node为每个阶段异步最大连接数,可配。

2、

1)先申请as_event_loop_capacity(为)个异步队列空间

2)将最大连接数分配到每个异步队列中。Max=最大连接/队列个数,rem=0

3、初始化每个队列。队列为空

4、

 

 

//1、

as_node_create:

if (as_event_loop_capacity > 0) {

         node->async_conn_qs=

as_node_create_async_queues(cluster->async_max_conns_per_node);

         node->pipe_conn_qs= as_node_create_async_queues(cluster->pipe_max_conns_per_node);

}

 

//2、

as_node_create_async_queues(uint32_tmax_conns_per_node):

as_queue* queues = cf_malloc(sizeof(as_queue)* as_event_loop_capacity);

uint32_t max = max_conns_per_node /as_event_loop_capacity;

uint32_t rem = max_conns_per_node - (max *as_event_loop_capacity);

uint32_t capacity;

for (uint32_t i = 0; i <as_event_loop_capacity; i++) {

         capacity= i < rem ? max + 1 : max;

         as_queue_init(&queues[i],sizeof(void*), capacity);

}

3、

Bool as_queue_init(as_queue* queue,uint32_t item_size, uint32_t capacity)

{

         queue->data= cf_malloc(capacity * item_size);

        

         if(! queue->data) {

                   returnfalse;

         }

         queue->capacity= capacity;//capacity为每个队列的容量

         queue->head= queue->tail = 0;//队列为空

         queue->item_size= item_size;//item大学

         queue->total= 0;

         queue->flags= ITEMS_ON_HEAP;

         returntrue;

}

 

4、异步,连接server之前,需要先创建eventloop。该函数入参即为as_event_loop_capacity。

  每个event_loop有一个as_ev_worker线程

as_event_create_loops

 

5、aerospike_key_put_async异步接口,有个入参指定event_loop,

   调用aerospike_key_put_async_ex

5.1、异步接口会调用as_async_write_command_create,拼接CMD

         该cmd->event_loop为接口中传进的值,如果不从接口传入,采用轮询方式分配

     cmd->event_loop= as_event_assign(event_loop);

                   returnevent_loop ? event_loop : as_event_loop_get();

5.2、异步接口拼接完cmd后,调用as_event_command_execute(cmd,err)去执行

5.3、as_event_command_execute执行方法:

         1)如果已经在该线程中,则直接执行

         2)否则保存到event_loop队列,并唤起event_loop线程进行处理

   

   if (cmd->event_loop->thread == pthread_self()) {

                   //We are already in event loop thread, so start PRocessing.

                   as_event_command_begin(cmd);

         }

         else{

                   if(cmd->timeout_ms) {

                            //Store current time in first 8 bytes which is not used yet.

                            *(uint64_t*)cmd= cf_getms();

                   }

                  

                   //Send command through queue so it can be executed in event loop thread.

                   if(! as_event_send(cmd)) {

                            as_event_command_free(cmd);

                            returnas_error_set_message(err, AEROSPIKE_ERR_CLIENT, "Failed to queuecommand");

                   }

         }

 

as_event_send

         as_event_loop*event_loop = cmd->event_loop;

         pthread_mutex_lock(&event_loop->lock);

         boolqueued = as_queue_push(&event_loop->queue, &cmd);

         pthread_mutex_unlock(&event_loop->lock);

         if(queued) {

                   ev_async_send(event_loop->loop,&event_loop->wakeup);

         }

 

6、执行event的函数as_event_command_begin

6.1、获取连接

         as_connection_statusstatus = cmd->pipe_listener != NULL ? as_pipe_get_connection(cmd) :as_event_get_connection(cmd);

6.2、执行

         if(status == AS_CONNECTION_FROM_POOL) {

                   as_ev_command_write_start(cmd);

         }

         elseif (status == AS_CONNECTION_NEW) {

                   as_ev_connect(cmd);//创建新连接后,进行连接

         }

6.3、获取连接的方法:

         1)as_queue*queue = &cmd->node->async_conn_qs[cmd->event_loop->index];得到异步连接队列

         2)如果队列不为空,则从中pop一个连接(怎么不使用锁保护?)

         3)否则新创建一个连接。这里有个限制。

                   每新创建一个连接queue->total++;

当queue->total>=queue->capicity的时候报连接耗尽错误。

6.4、

aerospike_key_put_async->aerospike_key_put_async_ex->as_async_write_command_create

中有个回调函数as_event_command_parse_header

6.5、该回调函数调用as_event_response_complete(cmd);将异步新创建的连接放到池子里

         as_event_response_complete->as_event_put_connection


发表评论 共有条评论
用户名: 密码:
验证码: 匿名发表