当前位置: > Linux服务器 > nginx >

nginx共享内存

时间:2016-06-06 11:56来源:linux.it.net.cn 作者:IT

 对于一个服务器模型来说,事件模型是至关重要的,nginx本身的高性能也归功于它的事件模型。一般来说,nginx的事件模型是基于epoll。而epoll中会调用3函数,epoll_create,epoll_ctl,epoll_wait.

   (1) 首先介绍一些相关的数据结构:

typedef struct {
    ngx_int_t  (*add)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);  //将某描述符的某个事件(可读/可写)添加到多路复用监控里
    ngx_int_t  (*del)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);  //将某描述符的某个事件(可读/可写)从多路复用监控里删除 

    ngx_int_t  (*enable)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);  //启动对某个事件的监控
    ngx_int_t  (*disable)(ngx_event_t *ev, ngx_int_t event, ngx_uint_t flags);  //禁止对某个事件的监控

    ngx_int_t  (*add_conn)(ngx_connection_t *c);    //将指定的连接关联的描述符添加到多路复用的监控里
    ngx_int_t  (*del_conn)(ngx_connection_t *c, ngx_uint_t flags);//将指定的连接关联的描述符从多路复用的监控里删除
    ngx_int_t (*process_changes)(ngx_cycle_t *cycle, ngx_uint_t nowait);//只有kqueue用到
    ngx_int_t (*process_events)(ngx_cycle_t *cycle, ngx_msec_t timer, ngx_uint_t flags);  //阻塞等待事件发生,对发生的事件进行逐个处理

    ngx_int_t (*init)(ngx_cycle_t *cycle, ngx_msec_t timer); //初始化 
    void (*done)(ngx_cycle_t *cycle);//回收资源
} ngx_event_actions_t; 
extern ngx_event_actions_t   ngx_event_actions;   //定义全局的 ngx_event_actions
#define ngx_process_changes  ngx_event_actions.process_changes
#define ngx_process_events   ngx_event_actions.process_events
#define ngx_done_events      ngx_event_actions.done

#define ngx_add_event        ngx_event_actions.add
#define ngx_del_event        ngx_event_actions.del
#define ngx_add_conn         ngx_event_actions.add_conn
#define ngx_del_conn         ngx_event_actions.del_conn
typedef struct {
    ngx_uint_t    connections;
    ngx_uint_t    use;

    ngx_flag_t    multi_accept;
    ngx_flag_t    accept_mutex;

    ngx_msec_t    accept_mutex_delay;

    u_char       *name;

#if (NGX_DEBUG)
    ngx_array_t   debug_connection;
#endif
} ngx_event_conf_t;


typedef struct {
    ngx_str_t              *name;

    void                 *(*create_conf)(ngx_cycle_t *cycle);
    char                 *(*init_conf)(ngx_cycle_t *cycle, void *conf);

    ngx_event_actions_t     actions;
} ngx_event_module_t;

    使用epoll事件模型时会调用的函数:

ngx_event_module_t  ngx_epoll_module_ctx = {
    &epoll_name,
    ngx_epoll_create_conf,               /* create configuration */
    ngx_epoll_init_conf,                 /* init configuration */

    {
        ngx_epoll_add_event,             /* add an event */   
        ngx_epoll_del_event,             /* delete an event */
        ngx_epoll_add_event,             /* enable an event */
        ngx_epoll_del_event,             /* disable an event */
        ngx_epoll_add_connection,        /* add an connection */
        ngx_epoll_del_connection,        /* delete an connection */
        NULL,                            /* process the changes */
        ngx_epoll_process_events,        /* process the events */
        ngx_epoll_init,                  /* init the events */
        ngx_epoll_done,                  /* done the events */
    }
};

 

(2)事件模型

      ngx_events_module是一个核心模块,  由它来完成event  module  的初始化.在obj/ngx_modules.c文件中我们只发现了2个event module。分别是ngx_event_core_module和ngx_epoll_module。ngx_event_core_module这个模块在事件模型初始化过程中起着至关重要的作用,而ngx_epoll_module实际上就是底层io模型的实现。事件模型的初始化与http模块类似,由ngx_events_module驱动整个事件模块的解析和初始化,ngx_event_core_module对events块大部分指令的解析保存重要的配置信息。

static ngx_command_t  ngx_events_commands[] = {

    { ngx_string("events"),
      NGX_MAIN_CONF|NGX_CONF_BLOCK|NGX_CONF_NOARGS,
      ngx_events_block,  //events模块的入口函数
      0,
      0,
      NULL },

      ngx_null_command
};

事件模型的初始化的入口函数:

  1.  ngx_events_block. 

static char *
ngx_events_block(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
{
    char                 *rv;
    void               ***ctx;
    ngx_uint_t            i;
    ngx_conf_t            pcf;
    ngx_event_module_t   *m;

    //为ctx分配空间,所有的event module的全局配置是一个数组,这里也为它分配空间,同时将存放在ngx_cycle->conf_ctx数组的ngx_events_module的配置结构赋值为ctx
    ctx = ngx_pcalloc(cf->pool, sizeof(void *));
    if (ctx == NULL) {
        return NGX_CONF_ERROR;
    }
    *ctx = ngx_pcalloc(cf->pool, ngx_event_max_module * sizeof(void *));
 
    //将在ngx_cycle->conf_ctx数组中的存放的ngx_events_moudle的config信息赋值给ctx,也就是event  module配置信息的数组
    *(void **) conf = ctx;

    for (i = 0; ngx_modules[i]; i++) {
        if (ngx_modules[i]->type != NGX_EVENT_MODULE) {
            continue;
        }

        m = ngx_modules[i]->ctx;
        //调用create_conf回调函数创建配置信息结构
        if (m->create_conf) {
            (*ctx)[ngx_modules[i]->ctx_index] = m->create_conf(cf->cycle);
            if ((*ctx)[ngx_modules[i]->ctx_index] == NULL) {
                return NGX_CONF_ERROR;
            }
        }
    }
    //为解析events块准备,设置解析模块的类型,  备份cf,解析完events块后恢复
    pcf = *cf;
    cf->ctx = ctx;
    cf->module_type = NGX_EVENT_MODULE;
    cf->cmd_type = NGX_EVENT_CONF;
    //解析events块
    rv = ngx_conf_parse(cf, NULL);  //递归调用,复杂块
    //恢复cf 
    *cf = pcf;

    if (rv != NGX_CONF_OK)
        return rv;

    for (i = 0; ngx_modules[i]; i++) {
        if (ngx_modules[i]->type != NGX_EVENT_MODULE) {
            continue;
        }

        m = ngx_modules[i]->ctx;
// 解析完events块,接着调用所有event module的init_conf回调函数初始化模块的配置结构。这里ngx_event_core_module和ngx_epoll_module会对配置结构中尚未初始化的一些属性赋默认值,比如默认使用io模型,也就是use指令的默认值。
        if (m->init_conf) {
            rv = m->init_conf(cf->cycle, (*ctx)[ngx_modules[i]->ctx_index]);
            if (rv != NGX_CONF_OK) {
                return rv;
            }
        }
    }

    return NGX_CONF_OK;
}

     看到这里,events块已经解析完毕,为什么还没看到epoll.在我们解析完配置文件之后会依次调用init_module和init_process,所以事件模型一定是在这两个时间点初始化的。下面看一下ngx_event_core_moduel的结构:

ngx_module_t  ngx_event_core_module = {
    NGX_MODULE_V1,
    &ngx_event_core_module_ctx,            /* module context */
    ngx_event_core_commands,               /* module directives */
    NGX_EVENT_MODULE,                      /* module type */
    NULL,                                  /* init master */
    ngx_event_module_init,                 /* init module */ //ngx_init_cycle中被调用,初始化共享内存中存放的数据结构,accept锁,链接计数器
    ngx_event_process_init,                /* init process *///在创建worker进程后调用
    NULL,                                  /* init thread */
    NULL,                                  /* exit thread */
    NULL,                                  /* exit process */
    NULL,                                  /* exit master */
    NGX_MODULE_V1_PADDING
};
init_module:
static ngx_int_t
ngx_event_module_init(ngx_cycle_t *cycle)
{
    .......
      cl = 128;

    size = cl            /* ngx_accept_mutex */
           + cl          /* ngx_connection_counter */
           + cl;         /* ngx_temp_number */

    shm.size = size;
    shm.name.len = sizeof("nginx_shared_zone");
    shm.name.data = (u_char *) "nginx_shared_zone";
    shm.log = cycle->log;

     if (ngx_shm_alloc(&shm) != NGX_OK) {
        return NGX_ERROR;
    }

    shared = shm.addr;

    ngx_accept_mutex_ptr = (ngx_atomic_t *) shared;
    ngx_accept_mutex.spin = (ngx_uint_t) -1;
    //  将accept锁放入共享内存,并将其初始化。
    if (ngx_shmtx_create(&ngx_accept_mutex, (ngx_shmtx_sh_t *) shared,
                         cycle->lock_file.data)
        != NGX_OK)
    {
        return NGX_ERROR;
    }

    ngx_connection_counter = (ngx_atomic_t *) (shared + 1 * cl);//接着放入连接计数器
}

init_module的回调函数执行完了,接下来执行init_process回调函数.

3.   ngx_event_process_init函数是在创建完worker进程后调用的,实际的初始化工作的大部分由它完成.

static ngx_int_t
ngx_event_process_init(ngx_cycle_t *cycle)
{
    //而对于ecf->accept_mutex字段的判断主要是提供用户便利,可以关闭该功能,因为既然均衡策略也有相应的代码逻辑,难保在某些情况下其本身的消耗也许会得不偿失;当然,该字段默认为1,在配置初始化函数ngx_event_core_init_conf()内,有这么一句:ngx_conf_init_value(ecf->accept_mutex, 1);
    if (ccf->master && ccf->worker_processes > 1 && ecf->accept_mutex) {  
    ngx_use_accept_mutex = 1;  
    ngx_accept_mutex_held = 0;  
    ngx_accept_mutex_delay = ecf->accept_mutex_delay;  
  
    } else {  
        ngx_use_accept_mutex = 0;  
    }  
}

根据配置信息初始化ngx_use_accept_mutex,这个变量用于指示是否使用accept锁.

//初始化timer
if (ngx_event_timer_init(cycle->log) == NGX_ERROR) {  
    return NGX_ERROR;  
}  
  for (m = 0; ngx_modules[m]; m++) {
        if (ngx_modules[m]->type != NGX_EVENT_MODULE) {
            continue;
        }

        if (ngx_modules[m]->ctx_index != ecf->use) {
            continue;
        }

        module = ngx_modules[m]->ctx;

        if (module->actions.init(cycle, ngx_timer_resolution) != NGX_OK) { //初始化epoll,在lunix下  ngx_epoll_init
            /* fatal */
            exit(2);
        }

        break;
    }

 这端代码完成了事件模型的初始化。遍历所有的事件模块找到通过use设定的事件模块,然后调用该模块init函数来初始化事件模型.默认nginx使用epoll,对应 的模块是ngx_epoll_module,它指定的init函数为ngx_epoll_init函数.该函数会调用epoll_create.

   4.ngx_epoll_init    

static ngx_int_t
ngx_epoll_init(ngx_cycle_t *cycle, ngx_msec_t timer)
{
       ep = epoll_create(cycle->connection_n / 2);//  ep就是epoll的句柄,初值为-1,所以一启动nginx就是调用epoll_create创建句柄,

         if (nevents < epcf->events) {
                  if (event_list) {
                          ngx_free(event_list);
         }

         //初始化nevents和event_list,epcf->events是由ngx_epoll_module的epoll_events指令设置的。nevents和event_list是要传给epoll_wait的参数,nevents是要监听的事件的最大个数,event_list用于存放epoll返回的事件。

         event_list = ngx_alloc(sizeof(struct epoll_event) * epcf->events,ycle->log);

         nevents = epcf->events;

         ngx_event_actions = ngx_epoll_module_ctx.actions; //为抽象事件模型赋值

}

 为ngx_event_actions赋值,之后ngx_event_actions指向epoll的事件结构。在此之后,所有的事件操作都可以由一组宏完成,定义在ngx_event.h:

#define ngx_process_changes  ngx_event_actions.process_changes  
#define ngx_process_events   ngx_event_actions.process_events  
#define ngx_done_events      ngx_event_actions.done  
  
#define ngx_add_event        ngx_event_actions.add  
#define ngx_del_event        ngx_event_actions.del  
#define ngx_add_conn         ngx_event_actions.add_conn  
#define ngx_del_conn         ngx_event_actions.del_conn  

在回到ngx_event_process_init函数 :

为nginx中的链接池分配存储空间,cycle->connection_n是由worker_connection指令设置的。
    cycle->connections =  ngx_alloc(sizeof(ngx_connection_t) * cycle->connection_n, cycle->log);
    if (cycle->connections == NULL) {
        return NGX_ERROR;
    }
    c = cycle->connections;
    //为读事件表分配空间并初始化
    cycle->read_events = ngx_alloc(sizeof(ngx_event_t) * cycle->connection_n,  cycle->log);  
    if (cycle->read_events == NULL) {  
        return NGX_ERROR;  
    }  
  
    rev = cycle->read_events;  
    for (i = 0; i < cycle->connection_n; i++) {  
        rev[i].closed = 1;  // 初始化是关闭的连接  
        rev[i].instance = 1;  
#if (NGX_THREADS)  
        rev[i].lock = &c[i].lock;  
        rev[i].own_lock = &c[i].lock;  
#endif  
    }  
   //为写事件表分配空间并初始化
    cycle->write_events = ngx_alloc(sizeof(ngx_event_t) * cycle->connection_n,
                                    cycle->log);
    if (cycle->write_events == NULL) {
        return NGX_ERROR;
    }

    wev = cycle->write_events;
 
   for (i = 0; i < cycle->connection_n; i++) {wev[i].closed = 1;}
i = cycle->connection_n;
    next = NULL;

    do {
        i--;

        c[i].data = next;
        c[i].read = &cycle->read_events[i];
        c[i].write = &cycle->write_events[i];
        c[i].fd = (ngx_socket_t) -1;

        next = &c[i];

#if (NGX_THREADS)
        c[i].lock = 0;
#endif
    } while (i);

初始化链接池,每个链接的读写事件在read_events和write_events数组中的下标和这个连接在connections数组中的下标是一致的。nginx将连接池组织成一种类似链表的结构,获取和 释放连接很方便.

//初始化连接池   一开始时链接池都是空闲的
cycle->free_connections = next;
cycle->free_connection_n = cycle->connection_n;
    ls = cycle->listening.elts;
    for (i = 0; i < cycle->listening.nelts; i++) {

        c = ngx_get_connection(ls[i].fd, cycle->log);

        if (c == NULL) {
            return NGX_ERROR;
        }

        c->log = &ls[i].log;

        c->listening = &ls[i];
        ls[i].connection = c;

        rev = c->read;

        rev->log = c->log;
        rev->accept = 1;// 将连接的accept设置成1,表示可以接受连接请求

#if (NGX_HAVE_DEFERRED_ACCEPT)
        rev->deferred_accept = ls[i].deferred_accept;
#endif

        if (!(ngx_event_flags & NGX_USE_IOCP_EVENT)) {
            if (ls[i].previous) {

                /*
                 * delete the old accept events that were bound to
                 * the old cycle read events array
                 */

                old = ls[i].previous->connection;

                if (ngx_del_event(old->read, NGX_READ_EVENT, NGX_CLOSE_EVENT)
                    == NGX_ERROR)
                {
                    return NGX_ERROR;
                }

                old->fd = (ngx_socket_t) -1;
            }
        }
        //在监听到连接的时候,调用ngx_event_accept函数 
        rev->handler = ngx_event_accept;//设置读事件的处理函数,上面已经设置了rev->accept = 1,也就是说这是监听套接字,
        //使用accept锁时,等worker进程抢到accept锁,再加入epoll事件循环中
        if (ngx_use_accept_mutex) {
            continue;
        }
        //在不使用accept锁时,直接将事件加入epoll事件循环
        if (ngx_event_flags & NGX_USE_RTSIG_EVENT) {
            if (ngx_add_conn(c) == NGX_ERROR) {
                return NGX_ERROR;
            }

        } else {
            if (ngx_add_event(rev, NGX_READ_EVENT, 0) == NGX_ERROR) {
                return NGX_ERROR;
            }
        }

       最后这部分代码完成的是最重要的功能,为所有监听socket分配连接并初始化。在不使用accept锁的情况会直接将所有的监听socket放入epoll事件循环,而在使用accept锁时worker进程必须获得锁后才能将监听socket加入事件循环,这部分工作在事件主循环中完成。上面就是nginx事件模型初始化的全部过程,我们看到了epoll_create和epoll_ctl,那么epoll_wait在那调用的呢?大家还记得在worker进程的主循环中调用的ngx_process_events_and_timers函数吧,之前介绍过这个函数是用于处理网络io事件的,实际就是epoll_wait被调用的地方。

     5. ngx_process_events_and_timers

 (ngx_use_accept_mutex) {
        if (ngx_accept_disabled > 0) {
            ngx_accept_disabled--;

        } else {
            if (ngx_trylock_accept_mutex(cycle) == NGX_ERROR) {
                return;
            }

            if (ngx_accept_mutex_held) {  //当前是否拥有锁
                flags |= NGX_POST_EVENTS;  //所有发生的事件都将延后处理,先释放锁

            } else {
                if (timer == NGX_TIMER_INFINITE
                    || timer > ngx_accept_mutex_delay)
                {
                    timer = ngx_accept_mutex_delay;
                }
            }
        }
    }

    在使用accept锁的情况下,需要先获得这个锁才能accept到新的连接,通过调用ngx_trylock_accept_mutex可以抢夺锁抢夺锁并accept连接。ngx_accept_disabled变量用于实现简单的负载均衡,防止连接分配的不均匀。这个变量在ngx_event_accept函数被设置为:ngx_cycle->connection_n/8 - ngx_cycle->free_connection_n,当ngx_accept_disabled大于0时,worker进程将放弃抢夺锁,专心处理已有的连接,并把ngx_accept_disabled变量减1。

     ngx_accept_mutex_held标记指示进程是否获得锁,得到锁的进程会添加NGX_POST_EVENTS标记,这个标记意味着将所有发生的事件放到一个队列中在,等到进程释放锁之后再慢慢处理,因为请求的处理可能非常耗时,如果不体现释放锁的话,会导致其他进程一直获取不到锁,这样accept的效率很低。在ngx_trylock_accept_mutex函数内部会先获取锁,然后再调用ngx_enable_accept_events把所有的监听socket添加到epoll事件循环中,下面看一下这个函数。

static ngx_int_t
ngx_enable_accept_events(ngx_cycle_t *cycle)
{
    ngx_uint_t         i;
    ngx_listening_t   *ls;
    ngx_connection_t  *c;

    ls = cycle->listening.elts;
    for (i = 0; i < cycle->listening.nelts; i++) {

        c = ls[i].connection;

        if (c->read->active) {
            continue;
        }

        if (ngx_event_flags & NGX_USE_RTSIG_EVENT) {

            if (ngx_add_conn(c) == NGX_ERROR) {
                return NGX_ERROR;
            }

        } else {
            if (ngx_add_event(c->read, NGX_READ_EVENT, 0) == NGX_ERROR) {
                return NGX_ERROR;
            }
        }
    }

    return NGX_OK;
}

在ngx_event_process_init中对于使用accept锁的情况,没有将监听socket加入epoll事件循环,而是这个函数中完成的。下面继续看ngx_process_events_and_timers函数。

delta = ngx_current_msec;
(void) ngx_process_events(cycle, timer, flags);
delta = ngx_current_msec - delta;

上面代码会调用ngx_process_events,根据上面的宏定义,实际会调用ngx_epoll_preocess_events,这个函数会调用epoll_wait监听事件 。

6. ngx_epoll_process_events

    events = epoll_wait(ep, event_list, (int) nevents, timer);

    err = (events == -1) ? ngx_errno : 0;

    if (flags & NGX_UPDATE_TIME || ngx_event_timer_alarm) {
        ngx_time_update();
    }

调用epoll_wait函数,根据flag标签更新时间.

for (i = 0; i < events; i++) {
        c = event_list[i].data.ptr;//事件对应的连接信息ngx_connection_t

        instance = (uintptr_t) c & 1;
        c = (ngx_connection_t *) ((uintptr_t) c & (uintptr_t) ~1);

        rev = c->read;

        revents = event_list[i].events; //发生的事件
        //处理读事件
        if ((revents & EPOLLIN) && rev->active) {

            if ((flags & NGX_POST_THREAD_EVENTS) && !rev->accept) {
                rev->posted_ready = 1;

            } else {
                rev->ready = 1;//epoll设置
            }

            if (flags & NGX_POST_EVENTS)  
   // * 使用accept锁,此处会将事件添加到队列中。 这里根据是不是监听socket放到不同的队列
                queue = (ngx_event_t **) (rev->accept ?  &ngx_posted_accept_events : &ngx_posted_events);
                 // 添加到队列 
                ngx_locked_post_event(rev, queue);

            } else {  //如果没有使用accept锁的话,则直接使用回调函数.
                rev->handler(rev);
            }
        }

        wev = c->write;
if ((revents & EPOLLOUT) && wev->active) {

            if (c->fd == -1 || wev->instance != instance) {

                /*
                 * the stale event from a file descriptor
                 * that was just closed in this iteration
                 */

                ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
                               "epoll: stale event %p", c);
                continue;
            }

            if (flags & NGX_POST_THREAD_EVENTS) {
                wev->posted_ready = 1;

            } else {
                wev->ready = 1;
            }

            if (flags & NGX_POST_EVENTS) {
                ngx_locked_post_event(wev, &ngx_posted_events);

            } else {
                wev->handler(wev);
            }
        }
}

    接下来遍历所有返回的事件对这些事件进行处理,和正常使用epoll差不多,关于读写事件的处理直接看注释。这就是ngx_epoll_process_events的全过程,下面继续介绍ngx_process_events_and_timers的后面流程。

  //ngx_posted_accept_events是accept事件的队列,这里会依次调用每个事件的handler并将其删除。accept事件的handler就是ngx_event_accept,它是在ngx_event_process_init中为每个监听socket的读事件添加的,用来处理新建连接的初始化
     if (ngx_posted_accept_events) {
        ngx_event_process_posted(cycle, &ngx_posted_accept_events);
    }
    //释放accept锁 
    if (ngx_accept_mutex_held) {
        ngx_shmtx_unlock(&ngx_accept_mutex);
    }

    if (delta) {
        ngx_event_expire_timers();
    }
if (ngx_posted_events) {
        if (ngx_threaded) {
            ngx_wakeup_worker_thread(cycle);

        } else {
            ngx_event_process_posted(cycle, &ngx_posted_events);
        }
    }

 




(责任编辑:IT)
------分隔线----------------------------
栏目列表
推荐内容