2014年12月15日月曜日

CORE出力フィルタ: ソケット送信処理(1)

CORE出力フィルタ(core_output_filter)から呼び出されているソケット送信処理を実行する関数は以下の2つだった。

(1) send_brigade_nonblocking
(2) send_brigade_blocking

これを追ってみる。

(1)send_brigade_nonblocking関数


bucket brigade bb の先頭から、データを持っているデータバケットを対象に送信を行う。
ソケット送信を実行すると、この関数は(未送信のバケットがあっても)復帰する。
送信済みのbucketは、bbから外され、破棄される。

データバケットはFILEバケットをEnableSendFile on でsendfile()で送信する場合は、
ファイルサイズが256バイト以上であれば、sendfile()で送信する。
sendfileで完全に送信が完了しなかった場合には、FILEバケットが分割され、未送信のFILEバケットが bucket brigade bb に残される。

それ以外のデータバケットの場合、writev()で送信する。この時データバケットごとに
I/Oベクトルの配列にデータをセットし、一定数を超えた場合にはソケット送信を実行する。

あるいは、bucket brigade bbのすべてを処理終えたした時にも、送信データがあればソケット送信を行う。

全データが送信できなかった場合には、送信済みのデータバケットは破棄され、未送信の
データバケットが bucket brigade bb に残される。

以下詳細を見てみる。

(httpd-2.4.9/server/core_filters.c)

    616 static apr_status_t send_brigade_nonblocking(apr_socket_t *s,
    617                                              apr_bucket_brigade *bb,
    618                                              apr_size_t *bytes_written,
    619                                              conn_rec *c)

apr_socket_t *s 通信用のソケット情報
apr_bucket_brigade *bb 送信データ用bucket brigade
apr_size_t *bytes_written 出力データサイズ(関数側で返す)
conn_rec *c コネクション情報

    620 {
    621     apr_bucket *bucket, *next;
    622     apr_status_t rv;
    623     struct iovec vec[MAX_IOVEC_TO_WRITE];
    624     apr_size_t nvec = 0;
    625
    626     remove_empty_buckets(bb);

remove_empty_buckets(bb)は、bucket brigade bbを先頭から辿り、
メタデータバケットかlength==0のデータバケットが続いている間、
そのバケットを bb から取り除き、破棄している。

length!=0のデータバケットが残った場合に、bucket brigadeが処理される。

    627
    628     for (bucket = APR_BRIGADE_FIRST(bb);
    629          bucket != APR_BRIGADE_SENTINEL(bb);
    630          bucket = next) {
    631         next = APR_BUCKET_NEXT(bucket);
    632 #if APR_HAS_SENDFILE

CentOSにはsendfileは存在する

    633         if (APR_BUCKET_IS_FILE(bucket)) {

FILEバケットの場合、
(1)EnableSendFileディレクティブが offではない場合
かつ
(2)サイズがAP_MIN_SENDFILE_BYTES(256)バイト以上の場合
sendfileを使った処理を行う。

EnableSendfiFileがonであっても小さなファイルの送信にはsendfileは使われないようだ。

    634             apr_bucket_file *file_bucket = (apr_bucket_file *)(bucket->data);
    635             apr_file_t *fd = file_bucket->fd;
    636             /* Use sendfile to send this file unless:
    637              *   - the platform doesn't support sendfile,
    638              *   - the file is too small for sendfile to be useful, or
    639              *   - sendfile is disabled in the httpd config via "EnableSendfile off"
    640              */
    641
    642             if ((apr_file_flags_get(fd) & APR_SENDFILE_ENABLED) &&
    643                 (bucket->length >= AP_MIN_SENDFILE_BYTES)) {


     56 #define AP_MIN_SENDFILE_BYTES           (256)


    644                 if (nvec > 0) {

nvecは初期値は0
bucket brigadeを処理していく中で、I/Oベクトルにデータを追加している。
そのバッファされているI/Oベクトルのデータを出力しておく。

    645                     (void)apr_socket_opt_set(s, APR_TCP_NOPUSH, 1);
    646                     rv = writev_nonblocking(s, vec, nvec, bb, bytes_written, c);
    647                     nvec = 0;
    648                     if (rv != APR_SUCCESS) {
    649                         (void)apr_socket_opt_set(s, APR_TCP_NOPUSH, 0);
    650                         return rv;
    651                     }
    652                 }
    653                 rv = sendfile_nonblocking(s, bucket, bytes_written, c);

FILEバケットをsendfileで送信する。
出力可能でなければ、ただちに復帰する。
ファイル全体を送信できた場合には、このbucketをbucket brigadeから取り除いて、apr_bucket_destroy()
を行う。
一部のみ送信できた場合は、このbucketを送信済みの部分と残った部分の2つに分割し、
送信済みのバケットを取り除き、apr_bucket_destroyを行う。


    654                 if (nvec > 0) {


ここには来ないのではないか(647行目で0になっている)。


    655                     (void)apr_socket_opt_set(s, APR_TCP_NOPUSH, 0);
    656                 }
    657                 if (rv != APR_SUCCESS) {
    658                     return rv;
    659                 }
    660                 break;


出力したら、処理を終える(forループを抜ける)。




    661             }
    662         }
    663 #endif /* APR_HAS_SENDFILE */
    664         /* didn't sendfile */
    665         if (!APR_BUCKET_IS_METADATA(bucket)) {

このif文はデータバケットの場合の処理

    666             const char *data;
    667             apr_size_t length;
    668
    669             /* Non-blocking read first, in case this is a morphing
    670              * bucket type. */
    671             rv = apr_bucket_read(bucket, &data, &length, APR_NONBLOCK_READ);

データバケットからデータを読み込む。
読込は非ブロックモードで行っている。

    672             if (APR_STATUS_IS_EAGAIN(rv)) {

上流から渡ってきているbucket brigadeにPIPEバケットやSOCKETバケットがあった場合、非ブロックで読み込んで、読み込みデータがなければ、この経路に入る。

    673                 /* Read would block; flush any pending data and retry. */
    674                 if (nvec) {

ここは、既にデータバケットを処理しており、未送信のものが存在している場合だ。

    675                     rv = writev_nonblocking(s, vec, nvec, bb, bytes_written, c);

I/Oベクトルのデータのソケット送信を実施する。

    676                     if (rv) {

ここは rv!=APR_SUCCESS(0)の場合の経路となる。

    677                         return rv;
    678                     }
    679                     nvec = 0;

nvecの値は0でリセットする。
この経路では、データがソケット送信された場合にも、引き続き処理が行われている。

    680                 }
    681
    682                 rv = apr_bucket_read(bucket, &data, &length, APR_BLOCK_READ);

そして、次にブロックモードでデータを読み込む。

    683             }
    684             if (rv != APR_SUCCESS) {

671行目のapr_bucket_read(APR_NONBLOCK_READ)の戻り値がEAGAIN以外のエラーか、
682行目のapr_bucket_read(APR_BLOCK_READ)がエラーの場合に
この経路に入る。

ここでは、エラーでreturn している。

    685                 return rv;
    686             }
    687
    688             /* reading may have split the bucket, so recompute next: */
    689             next = APR_BUCKET_NEXT(bucket);
    690             vec[nvec].iov_base = (char *)data;
    691             vec[nvec].iov_len = length;
    692             nvec++;

読み込んだデータをI/Oベクトルに追加する。

    693             if (nvec == MAX_IOVEC_TO_WRITE) {

I/Oベクトルの要素数がMAX_IOVEC_TO_WRITEになったら、出力を実行する。

    694                 rv = writev_nonblocking(s, vec, nvec, bb, bytes_written, c);

writev()関数を使って、I/Oベクトルの情報を出力する。
出力可能でなけばただちに復帰する。
出力完了したデータサイズに応じて、bucket brugade bb のlengthを消化して、
bucketをbb から取り除き、apr_bucket_destroy()を行う。
bucketのデータの一部しか送信できていない場合には、そのbucketを分割し、
送信済みのbucketを取り外して、apr_bucket_destroy()を行っている。

    695                 nvec = 0;

nvecは出力後、0にリセットする。

    696                 if (rv != APR_SUCCESS) {
    697                     return rv;
    698                 }
    699                 break;

出力したら、処理を終える(forループを抜ける)。

    700             }
    701         }
    702     }
    703
    704     if (nvec > 0) {

bucket brigadeを最後までチェックして、MAX_IOVEC_TO_WRITEにならなかった場合に
ここで出力する。

    705         rv = writev_nonblocking(s, vec, nvec, bb, bytes_written, c);
    706         if (rv != APR_SUCCESS) {
    707             return rv;
    708         }
    709     }
    710
    711     remove_empty_buckets(bb);
    712
    713     return APR_SUCCESS;
    714 }


(2)send_brigade_blocking関数


送信処理だが、こちらはブロックモードだ。
実装では、send_brigade_nonblocking()を呼んでいる。

ブロックモードでは、send_brigade_nonblocking()を実行して、結果がEAGAINの場合(書込みできない)、apr_socket_t情報のtimeoutの設定値を取得して、poll()で書込み可能になるまで待機する。

timeoutになる前に書込み可能になれば、再度send_brigade_nonblocking()を実行する。
timeoutした場合は、APR_TIMEUPでreturnする。
その他のエラーの場合もエラーでreturnする。

つまり、send_brigade_nonblockingと異なり、非ブロックで書き込みを行い、ソケットの書込み準備ができていない場合には、書込み可能になるまで、Timeoutディレクティブ等で指定されているタイムアウト値に従って待機するようになっている。

    726 static apr_status_t send_brigade_blocking(apr_socket_t *s,
    727                                           apr_bucket_brigade *bb,
    728                                           apr_size_t *bytes_written,
    729                                           conn_rec *c)
    730 {
    731     apr_status_t rv;
    732
    733     rv = APR_SUCCESS;
    734     while (!APR_BRIGADE_EMPTY(bb)) {
    735         rv = send_brigade_nonblocking(s, bb, bytes_written, c);

ここでsend_brigade_nonblocking()が実行されている。

    736         if (rv != APR_SUCCESS) {
    737             if (APR_STATUS_IS_EAGAIN(rv)) {

ソケットの書込み準備ができていなかった場合にこの経路に入る。


    738                 /* Wait until we can send more data */
    739                 apr_int32_t nsds;
    740                 apr_interval_time_t timeout;
    741                 apr_pollfd_t pollset;
    742
    743                 pollset.p = c->pool;
    744                 pollset.desc_type = APR_POLL_SOCKET;
    745                 pollset.reqevents = APR_POLLOUT;
    746                 pollset.desc.s = s;
    747                 apr_socket_timeout_get(s, &timeout);

apr_socket_t情報のtimeout設定を取得している。

    748                 do {
    749                     rv = apr_poll(&pollset, 1, &nsds, timeout);

ソケットが書き込み可能になるまで、所定の時間待つ。
シグナルで割りこまれた場合(EINTR)には処理を継続する。

    750                 } while (APR_STATUS_IS_EINTR(rv));
    751                 if (rv != APR_SUCCESS) {

apr_poll()でEINTR以外の何らかのエラーが発生したか、タイムアウトの場合の経路だ。

    752                     break;
    753                 }
    754             }
    755             else {

send_brigade_nonblockingがEAGAIN以外のエラーだった場合の経路になる

    756                 break;
    757             }
    758         }

send_brigade_nonblocking()が成功した場合、あるいは、ソケットが書き込み可能になった場合、
bucket brigadeが空になるまでsend_brigade_nonblocking()処理を継続する。

    759     }

send_brigade_nonblocking処理は、送信済みのbucketを破棄するので、すべて送信し終えれば、bucket brigadeが空となり
この734行からのwhile()ループを抜ける。

    760     return rv;
    761 }

以上、2関数を見たが、ここでも最終的なソケット送信関数は呼ばれていなかった。
ここでは送信処理として次の関数から呼ばれてる。

  • sendfile_nonblocking
  • writev_nonblocking

実際のところ、上記関数の内部を見ても、ソケット送信関数は呼ばれていない。
それぞれは、さらに次のAPRライブラリ関数を使用し、そこでようやく直接ソケットディスクリプタを扱う関数にたどり着く。

  • apr_socket_sendfile
  • apr_socket_sendv

もう少しなので、辿ってみることにする。

2014年12月8日月曜日

入力フィルタ: CORE_IN入力フィルタ

CORE_IN入力フィルタ: ap_core_input_filter


CORE_IN入力フィルタは、入力フィルタの最下層のネットワークフィルタになる。
この下には入力フィルタはないので、CIRE_IN入力フィルタでは、ネットワークからクライアントの送ってきたデータを受信し、処理して、上位の入力フィルタに引き渡す処理が行われる。
以前、入力フィルタの概要で少し見たが、今回はもう少し細かく全体を見ていく。

     94 apr_status_t ap_core_input_filter(ap_filter_t *f, apr_bucket_brigade *b,
     95                                   ap_input_mode_t mode, apr_read_type_e block,
     96                                   apr_off_t readbytes)

引数は次の通り。

引数 説明
ap_filter_t *f 入力フィルタ情報。fはap_core_input_filterに関する情報を格納する。
apr_bucket_brigade *b 吸い上げたデータを格納する宛先のbucket brigade。
ap_input_mode_t mode 入力モード。
AP_MODE_READBYTES 入力フィルタは最大readbytesバイト読み込んでreturn する
AP_MODE_GETLINE 入力フィルタは、1行(CRLF)分のデータを読み込んでreturnする。(1行が長すぎたり、あるいは、CRLFがないような場合には部分的なデータを読み込んでreturnする可能性がある)
AP_MODE_EATCRLF 入力フィルタは、CRLFを見つけたら消費(除去)する
AP_MODE_SPECULATIVE 入力フィルタの読み込みは投機的なものとみなされる。読み込まれたデータは、以降の別のモードでの処理のために蓄積される。
AP_MODE_EXHAUSTIVE 入力フィルタの読み込みは完全なものとみなされる。それ以上読込できなくなるまで読込が行われる。このモードは特に注意して扱うこと。
AP_MODE_INIT 入力フィルタはコネクションを初期化する。NNTP over SSL やFTP over SSL等で必要。
apr_read_type_e block 読込タイプ
APR_BLOCK_READ ブロックモード
APR_NONBLOCK_READ 非ブロックモード
apr_off_t readbytes 入力データサイズ


2014年12月1日月曜日

apr_socket_t 情報と SOCKETデータバケット

(1)SOCKETデータバケットタイプ情報

SOCKETデータバケットのバケットタイプ情報は次の通り。
(apr-util-1.5.4/buckets/apr_buckets_socket.c)

    107 APU_DECLARE_DATA const apr_bucket_type_t apr_bucket_type_socket = {
    108     "SOCKET", 5, APR_BUCKET_DATA,
    109     apr_bucket_destroy_noop,
    110     socket_bucket_read,
    111     apr_bucket_setaside_notimpl,
    112     apr_bucket_split_notimpl,
    113     apr_bucket_copy_notimpl
    114 };

(2)apr_socket_t 情報のセットアップ


SOCKETデータバケットで、独自の実装を持っているのはread関数だけだ。
それをみておきたいが、その前に、ソケット情報(apr_socket_t)を生成する listenerスレッドのacceptの処理部分を見たい。
listenerスレッドのListenソケットの処理ではacceptの処理について触れた。


(httpd-2.4.9/server/mpm/event/event.c)
   1368 static void * APR_THREAD_FUNC listener_thread(apr_thread_t * thd, void *dummy)
   1369 {
    :
   1616                     rc = lr->accept_func(&csd, lr, ptrans);

accept_func()の実体は、ap_unixd_accept()だ。
(httpd-2.4.9/server/mpm/event/event.c)
   1122 static apr_status_t init_pollset(apr_pool_t *p)
   1123 {
    :
   1151         lr->accept_func = ap_unixd_accept;

ap_unixd_accept()関数で、ソケットがacceptされ、apr_socket_t情報が作成される。
このapr_socket_t情報は、listenerスレッドの、void*型の変数csdにアドレスがセットされる。
(httpd-2.4.9/os/unix/unixd.c)
    291 AP_DECLARE(apr_status_t) ap_unixd_accept(void **accepted, ap_listen_rec *lr,
    292                                          apr_pool_t *ptrans)
    293 {
    294     apr_socket_t *csd;
    :
    300     *accepted = NULL;
    301     status = apr_socket_accept(&csd, lr->sd, ptrans);
    302     if (status == APR_SUCCESS) {
    303         *accepted = csd;
この*acceptedが、listenerスレッドの処理のcsdとなっている。


apr_socket_t情報は以下の通り。
(apr-1.5.1/include/arch/unix/apr_arch_networkio.h)

    103 struct apr_socket_t {
    104     apr_pool_t *pool;
    105     int socketdes;
    106     int type;
    107     int protocol;
    108     apr_sockaddr_t *local_addr;
    109     apr_sockaddr_t *remote_addr;
    110     apr_interval_time_t timeout;
    111 #ifndef HAVE_POLL
     :
    113 #endif
    114     int local_port_unknown;
    115     int local_interface_unknown;
    116     int remote_addr_unknown;
    117     apr_int32_t options;
    118     apr_int32_t inherit;
    119     sock_userdata_t *userdata;
    120 #ifndef WAITIO_USES_POLL
     :
    123 #endif
    124 };

もう少し先を辿る。
apr_socket_accept()だ。

2014年11月25日火曜日

apr_bucket と apr_bucket_brigade

今度は CORE_IN入力フィルタを見ようとしたところだが、入出力フィルタではbucketとbucket brigadeの処理が重要だ。
そこで、先に少し見ておくことにした。

bucket brigade(apr_bucket_brigade)は内部にbucket(apr_bucket)の列のリング構造を持っている。
bucket情報は前後へのポインタを持っており、bucket brigade情報自体もbucket情報へのポインタを持っている。
bucket brigadeの持っている次へのポインタが、bucket列の先頭に繋がり、bucket列を順に辿って、bucketの末尾の次が再び、bucket brigadeに繋がってリング構造となる。
逆方向も同じで、bucket brigadeの前にbucket列の末尾が繋がり、前に辿っていくと、bucket列の先頭に繋がり、その前に再びbucket brigadeが位置する。
このbucket brigade自身(内部)のアドレスは有効なbucket情報ではなく、センチネル(番兵)として機能する。



brigadeが軍隊がらみの用語らしく、bucket brigadeが何を意味するのか分からなかったが、単純に "bucket brigade" で辞書を引けば載っていた。バケツリレーの列のことだった。
バケツリレーか。なるほどね。

2014年11月17日月曜日

出力フィルタ: CORE出力フィルタ

CORE出力フィルタ: ap_core_output_filter



以前、出力フィルタの概要を見た。
今度は、出力フィルタの階層の最下流にあるCORE出力フィルタを少し細かく見る。
中間層の出力フィルタは上流から受け取った bucket brigadeを加工し、新しいbucket brigadeを作成すると、これを下流の出力フィルタに渡した。
最下流では、もう渡し先の出力フィルタはないので、データをクライアントに向けて送信する。


(httpd-2.4.9/server/core_filters.c)

    372 apr_status_t ap_core_output_filter(ap_filter_t *f, apr_bucket_brigade *new_bb)


ap_filter_t *fで、この出力フィルタ(ap_core_output_filter)のフィルタ情報が渡され、
apr_bucket_brigade *new_bbには送信するbacket brigadeが渡される。

    373 {
    374     conn_rec *c = f->c;

この処理のconn_rec情報。

    375     core_net_rec *net = f->ctx;

core_net_rec情報は、process_socket()で実行されるpre_connectionフック関数のうち
core_pre_connection()で用意される。

    376     core_output_filter_ctx_t *ctx = net->out_ctx;

CORE出力フィルタ用のコンテキスト情報。最初の関数呼び出しで本フィルタ内で作成される。

    377     apr_bucket_brigade *bb = NULL;
    378     apr_bucket *bucket, *next, *flush_upto = NULL;
    379     apr_size_t bytes_in_brigade, non_file_bytes_in_brigade;
    380     int eor_buckets_in_brigade, morphing_bucket_in_brigade;
    381     apr_status_t rv;
    382
    383     /* Fail quickly if the connection has already been aborted. */
    384     if (c->aborted) {

ここは異常系の処理だ。

    385         if (new_bb != NULL) {
    386             apr_brigade_cleanup(new_bb);
    387         }
    388         return APR_ECONNABORTED;
    389     }
    390
    391     if (ctx == NULL) {

コンテキスト情報 core_output_filter_ctx_t を用意する。
リクエスト処理で初めてCORE出力フィルタが呼ばれたときに行われることになる。

CORE出力フィルタのコンテキスト情報の定義は次の通り。

     81 struct core_output_filter_ctx {
     82     apr_bucket_brigade *buffered_bb;
     83     apr_bucket_brigade *tmp_flush_bb;
     84     apr_pool_t *deferred_write_pool;
     85     apr_size_t bytes_written;
     86 };

    392         ctx = apr_pcalloc(c->pool, sizeof(*ctx));
    393         net->out_ctx = (core_output_filter_ctx_t *)ctx;
    394         /*
    395          * Need to create tmp brigade with correct lifetime. Passing
    396          * NULL to apr_brigade_split_ex would result in a brigade
    397          * allocated from bb->pool which might be wrong.
    398          */

2つの作業用のbucket brigadeを作成し、コンテクスト情報に登録する。
いずれも、conn_recのプール(transactionプール)から確保している。

    399         ctx->tmp_flush_bb = apr_brigade_create(c->pool, c->bucket_alloc);
    400         /* same for buffered_bb and ap_save_brigade */
    401         ctx->buffered_bb = apr_brigade_create(c->pool, c->bucket_alloc);
    402     }
    403
    404     if (new_bb != NULL)
    405         bb = new_bb;

上流から渡されてきたbucket brigadeのアドレスを作業用の変数 bbにセットする。

    406
    407     if ((ctx->buffered_bb != NULL) &&
    408         !APR_BRIGADE_EMPTY(ctx->buffered_bb)) {
    409         if (new_bb != NULL) {
    410             APR_BRIGADE_PREPEND(bb, ctx->buffered_bb);
    411         }
    412         else {
    413             bb = ctx->buffered_bb;
    414         }

コンテキスト情報のbuffered_bbの内容と、今回の呼び出しで渡されてきた bucket brigadeの内容を
マージしている。
つまり、受け取ったnew_bbの前にフィルタの内部に前の処理時に留保されたbucket brigadeであるbuffered_bbを繋いでいる。

    415         c->data_in_output_filters = 0;

data_in_output_filtersは出力フィルタに未出力のデータ(bucket brigade)があるかどうかを判定している。
初期値として、なし(0)をセットする。

    416     }
    417     else if (new_bb == NULL) {

上流からbucket brigadeが渡されていない場合、
そして、buffered_bbにもデータがない場合、
処理対象となるbucketがないため、処理は行われず、直ちに成功でreturnする。

    418         return APR_SUCCESS;
    419     }

この部分のソースコードには、処理概要を説明している長めのコメントがある。
これを読めば、やっていることが分かるが、
ここでは、ソースコードに従って見ていきたいので、削った。
同じことを以下で書いている。
1)、2)だのa)、b)だのは、このコメントに書かれていた番号を該当する箇所の説明につけたものだ。

    467
    468     if (new_bb == NULL) {

1) ここは、コンテキスト情報のbufferd_bbに既にあったbucketだけが処理対象となるケース。

new_bbがNULLになるケースがどれほどあるか把握していないが、
process_socket()でCONN_STATE_WRITE_COMPLETION状態の場合に実行される
以下の989行目のコードは該当している。

(httpd-2.4.9/server/mpm/event/event.c)
    982     if (cs->pub.state == CONN_STATE_WRITE_COMPLETION) {
    983         ap_filter_t *output_filter = c->output_filters;
    984         apr_status_t rv;
    985         ap_update_child_status_from_conn(sbh, SERVER_BUSY_WRITE, c);
    986         while (output_filter->next != NULL) {
    987             output_filter = output_filter->next;
    988         }
    989         rv = output_filter->frec->filter_func.out_func(output_filter, NULL);

ここでは、無条件にデータの送信が非ブロックモードで実行される。

    469         rv = send_brigade_nonblocking(net->client_socket, bb,
    470                                       &(ctx->bytes_written), c);

send_brigade_nonblocking()処理内を辿ると、実際のクライアントへの送信処理が行われている。
net->client_socketはクライアントとの通信ソケット情報。
bbはbucket brigade。
ctx->bytes_writtenには送信済みデータ長が格納される(加算される)。
cはconn_rec情報。

    471         if (APR_STATUS_IS_EAGAIN(rv)) {
    472             rv = APR_SUCCESS;
    473         }
    474         else if (rv != APR_SUCCESS) {
    475             /* The client has aborted the connection */
    476             ap_log_cerror(APLOG_MARK, APLOG_TRACE1, rv, c,
    477                           "core_output_filter: writing data to the network");
    478             c->aborted = 1;
    479         }
    480         setaside_remaining_output(f, ctx, bb, c);

setaside_remaining_output()は、bucket brigade bbの未出力のデータを buffere_bbに退避する。
未出力データがあった場合には、識別用にconn_recのdata_in_output_filtersに1がセットされる。

送信処理が実行されると、このCORE出力関数からは復帰する。

    481         return rv;
    482     }

ここまでが1)の処理となる。

以下でいくつかの送信判定用の変数値が初期化される。

    483
    484     bytes_in_brigade = 0;
    485     non_file_bytes_in_brigade = 0;
    486     eor_buckets_in_brigade = 0;
    487     morphing_bucket_in_brigade = 0;

bytes_in_brigade bucket brigade内のデータバケットの lengthの合計を格納する(lengthが-1のものを除く)
non_file_bytes_in_brigade bucket brigade内のデータバケット(FILEを除く)の lengthの合計を格納する(lengthが-1のものを除く)
eor_buckets_in_brigade EORメタデータバケットの数をカウントする
morphing_bucket_in_brigade bucket brigade内のデータバケットでlengthが -1のものがあった場合に1を立てるフラグ


    488
    489     for (bucket = APR_BRIGADE_FIRST(bb); bucket != APR_BRIGADE_SENTINEL(bb);
    490          bucket = next) {
    491         next = APR_BUCKET_NEXT(bucket);

ここで、bufferd_bbとnew_bbが連結された bucket brigade bb 内のbucketが順に処理される。

    492
    493         if (!APR_BUCKET_IS_METADATA(bucket)) {

bucketがデータタイプのバケットの場合

    494             if (bucket->length == (apr_size_t)-1) {

これはバケットのデータ長が事前には不明な場合で、
たとえば、SOCKETバケットやPIPEバケットなど、
読み込んでみないとデータ長が分からないバケットが該当する。

    495                 /*
    496                  * A setaside of morphing buckets would read everything into
    497                  * memory. Instead, we will flush everything up to and
    498                  * including this bucket.
    499                  */
    500                 morphing_bucket_in_brigade = 1;
    501             }
    502             else {

バケットのデータ長が分かっている場合は、bytes_in_brigadeに値を足しこむ。

    503                 bytes_in_brigade += bucket->length;
    504                 if (!APR_BUCKET_IS_FILE(bucket))

特にバケットがFILEバケットでない場合に、non_file_bytes_in_brigadeに値を足しこむ

    505                     non_file_bytes_in_brigade += bucket->length;
    506             }
    507         }
    508         else if (AP_BUCKET_IS_EOR(bucket)) {

bucketが EORメタデータバケットの場合
eor_buckets_in_brigadeをカウントアップする。

    509             eor_buckets_in_brigade++;
    510         }
    511

2) 上記のbucketのチェックを終えて、ここまでのデータ(bucket列)を実際に送信するかどうかの判定を行う。

    512         if (APR_BUCKET_IS_FLUSH(bucket)
    513             || non_file_bytes_in_brigade >= THRESHOLD_MAX_BUFFER
    514             || morphing_bucket_in_brigade
    515             || eor_buckets_in_brigade > MAX_REQUESTS_IN_PIPELINE) {
    516             /* this segment of the brigade MUST be sent before returning. */

このif文の条件を満たすと、ここまでのbucketの送信を実行する。
下記でflush_uptoにセットされるnextが送信停止位置を示すことになる。
条件は以下のいずれかが真の場合だ。

(a)FLUSHメタデータバケットがあった場合
(b)FILEではないデータバケットのlengthの合計(non_file_bytes_in_brigade)がTHRESHOLD_MAX_BUFFER(64KB)以上になった場合

    364 #define THRESHOLD_MAX_BUFFER 65536

(d)データバケットにlengthが-1のものがあった場合(morphing_bucket_in_brigade==1)
(c)EORメタデータバケットがMAX_REQUESTS_IN_PIPELINE(5)個以上処理された場合

(eor_buckets_in_brigade)
    365 #define MAX_REQUESTS_IN_PIPELINE 5


    517
    518             if (APLOGctrace6(c)) {
    519                 char *reason = APR_BUCKET_IS_FLUSH(bucket) ?
    520                                "FLUSH bucket" :
    521                                (non_file_bytes_in_brigade >= THRESHOLD_MAX_BUFFER) ?
    522                                "THRESHOLD_MAX_BUFFER" :
    523                                morphing_bucket_in_brigade ? "morphing bucket" :
    524                                "MAX_REQUESTS_IN_PIPELINE";
    525                 ap_log_cerror(APLOG_MARK, APLOG_TRACE6, 0, c,
    526                               "core_output_filter: flushing because of %s",
    527                               reason);
    528             }
    529             /*
    530              * Defer the actual blocking write to avoid doing many writes.
    531              */
    532             flush_upto = next;

flush_uptoに値がセットされた場合、以降の処理で、先頭からこの直前(ここで処理されているbucket)までのbucket列を出力することになる(3)。

ここで、いったん送信条件をリセットする。

    533
    534             bytes_in_brigade = 0;
    535             non_file_bytes_in_brigade = 0;
    536             eor_buckets_in_brigade = 0;
    537             morphing_bucket_in_brigade = 0;
    538         }
    539     }
    540
    541     if (flush_upto != NULL) {

3) flush_uptoがNULLでない場合、ここまでのデータ(bucket列)の送信が実行される。

    542         ctx->tmp_flush_bb = apr_brigade_split_ex(bb, flush_upto,
    543                                                  ctx->tmp_flush_bb);

まず、flush_uptoから末尾までのbucket列を分割し、コンテキスト情報のtmp_flush_bbに退避する。

    544         rv = send_brigade_blocking(net->client_socket, bb,
    545                                    &(ctx->bytes_written), c);

そして、残った先頭からfulsh_uptoの前までのbucketからなる bucket brigade bb をブロックモードで送信する。

ブロックモードでの送信処理を辿ると、その内部で非ブロックモードの送信処理send_brigade_nonblocking()を呼び出している。
非ブロックでの送信処理の実行結果が、エラーで、エラーコードがEAGAIN/EWOULDBLOCK の場合に、poll関数で、書込み可能になるまで待機する(Timeoutの設定が待ち時間の上限)処理が追加されている。

    546         if (rv != APR_SUCCESS) {
    547             /* The client has aborted the connection */
    548             ap_log_cerror(APLOG_MARK, APLOG_TRACE1, rv, c,
    549                           "core_output_filter: writing data to the network");
    550             c->aborted = 1;
    551             return rv;
    552         }
    553         APR_BRIGADE_CONCAT(bb, ctx->tmp_flush_bb);

送信処理を終えると、tmp_flush_bbに退避しておいた残りのbucket列を bb に繋ぎなおしている。

    554     }
    555
    556     if (bytes_in_brigade >= THRESHOLD_MIN_WRITE) {

4) FILEを含むデータバケットのlengthの合計が、THRESHOLD_MIN_WRITE(4KB)以上になった場合にも出力処理を実行する。

    363 #define THRESHOLD_MIN_WRITE 4096


ここでは、出力は非ブロックモードで行われる。

    557         rv = send_brigade_nonblocking(net->client_socket, bb,
    558                                       &(ctx->bytes_written), c);
    559         if ((rv != APR_SUCCESS) && (!APR_STATUS_IS_EAGAIN(rv))) {
    560             /* The client has aborted the connection */
    561             ap_log_cerror(APLOG_MARK, APLOG_TRACE1, rv, c,
    562                           "core_output_filter: writing data to the network");
    563             c->aborted = 1;
    564             return rv;
    565         }
    566     }
    567


    568     setaside_remaining_output(f, ctx, bb, c);

未送信データが存在した場合buffered_bbに保持しておいて、識別用にconn_recのdata_in_output_filtersに1をセットする。

    569     return APR_SUCCESS;
    570 }


ここまでが、CORE出力関数となる。
実際のソケットの送信処理は、下記の関数内で実行されている。
いずれも server/core_filters.c ファイル内でstatic関数として定義されている。
概要は上述した通り。

  • send_brigade_nonblocking
  • send_brigade_blocking

送信の実体は、sendfile()であったり、バケットのデータをI/Oベクトルに収めてのwritev()であったりする。
これは、別途コードを追う予定。

2014年11月4日火曜日

リクエスト処理の流れ: eventMPM コネクション処理フロー

コネクション処理フロー

worker_thread()、process_socket()とlistener_thread()の処理を概観して、コネクションのconn_state_e情報と処理の流れを図にした。



accept待ちのListenソケットがlistener_threadで接続を受け付けて、workerスレッドのprocess_socketでのリクエスト処理/出力処理を経て、ソケットを閉じたり、次リクエスト受け付けを待つという流れになっている。
青い四角がworkerスレッドの処理、黄色い四角がlistenerスレッドの処理。ソケットクローズは、listenerスレッドで行われる。ただ、これは正常系の場合で、既に見てきているように、ソケットのクローズは、エラー時などには別のタイミングでも実行されている。

SUSPENDED状態を持つタイムアウト処理は、いまのところmod_dialupにしか実装は見当たらない。
タイムアウト処理をSUSPENDED状態の中で繰り返し、処理を終えると、ソケットを閉じる処理に向かっている。


2014年10月27日月曜日

リクエスト処理の流れ: listener_thread(5) タイムアウトキューの処理

(14.4) イベント処理のつづき。

(14.5) タイムアウトキューの処理


次は、4つ用意されているタイムアウトキューに対する、タイムアウト処理だ。
タイムアウトに至る前にイベントが発生して、1655行目までのソケットイベント処理で消化されれば
特に何も行われない。
タイムアウトに至るまでに、ソケットの読み込み/書き出しが可能とならなかった場合に、それらソケットが次の処理の対象となる。

   1656
   1657         /* XXX possible optimization: stash the current time for use as
   1658          * r->request_time for new requests
   1659          */
   1660         now = apr_time_now();

まず、現在時刻を取得する。単位はマイクロ秒だ。

   1661         /* we only do this once per 0.1s (TIMEOUT_FUDGE_FACTOR) */
   1662         if (now > timeout_time) {

timeout_timeは初期値は0なので、最初はこのif文に必ず入る。
そして、timeout_timeに現在時刻のTIMEOUT_FUDGE_FACTOR後にセットしている(1664行目)。
定義を見ると、100000マイクロ秒なので、0.1秒。

   1396 #define TIMEOUT_FUDGE_FACTOR 100000

おおむね、0.1秒以上間隔をあけてチェックするということだろう。

   1663             struct process_score *ps;
   1664             timeout_time = now + TIMEOUT_FUDGE_FACTOR;
   1665
   1666             /* handle timed out sockets */
   1667             apr_thread_mutex_lock(timeout_mutex);
   1668
   1669             /* Step 1: keepalive timeouts */
   1670             /* If all workers are busy, we kill older keep-alive connections so that they
   1671              * may connect to another process.
   1672              */

(1)KeepAliveタイムアウト

keepalive_qキューを処理する。

   1673             if (workers_were_busy && keepalive_q.count) {

この時点でworkers_were_busyが1の場合(アイドルworkerスレッドがいない場合)で、
KeepAliveタイムアウトキューに登録がある場合(次のリクエストを待っている場合)、
現在時刻(+0.1秒)からKeepAliveTimeOut先までのタイムアウトイベントを先取りして
リンガリングクローズを実行している。
上記のコメントによれば、すべてのworkerがビジーなら、古いKeep-Alive接続を終了し、
クライアントが必要になった時点で別のプロセスに接続できるようにするということだ。

しかし、KeepAliveTimeOut先でこのkeepalive_qキューのタイムアウトは登録されるのだから、
古いというか、おそらくこの時点で登録済みのすべてのKeepAlive待ちイベントが
全てstart_lingering_close_nonblocking()関数処理されると考えられる。

start_lingering_close_nonblocking()関数は、
既に見た
start_lingering_close_blocking
と異なり、shutdown(SHUT_WR)の前に未出力のデータのフラッシュを試みない。
それ以外は同様で、この処理によって、ソケットはLINGER待ちになりlinger_qキュー、またはshort_linger_qキューに
登録される。

   1674                 ap_log_error(APLOG_MARK, APLOG_TRACE1, 0, ap_server_conf,
   1675                              "All workers are busy, will close %d keep-alive "
   1676                              "connections",
   1677                              keepalive_q.count);
   1678                 process_timeout_queue(&keepalive_q,
   1679                                       timeout_time + ap_server_conf->keep_alive_timeout,
   1680                                       start_lingering_close_nonblocking);
   1681             }
   1682             else {

こちらは、通常の場合で、
timeout_time(現在時刻+0.1秒)時点でタイムアウトしているタイムアウトイベントを
同じく、start_lingering_close_nonblocking()関数処理している。


   1683                 process_timeout_queue(&keepalive_q, timeout_time,
   1684                                       start_lingering_close_nonblocking);
   1685             }
   1686             /* Step 2: write completion timeouts */

(2)書き込み完了タイムアウト


ここは、write_completion_qキューを処理する。
タイムアウト時間が来ていた場合に行う処理は、start_lingering_close_nonblocking()で、
これはKeepAlive待ちと同じだ。
通常は、shutdown(SHUT_WR)が行われ、linger_qキューに登録される。

   1687             process_timeout_queue(&write_completion_q, timeout_time,
   1688                                   start_lingering_close_nonblocking);
   1689             /* Step 3: (normal) lingering close completion timeouts */

(3)リンガリングクローズタイムアウト(標準)

linger_qキューの処理だ。
ここでのタイムアウト時の処理は、stop_lingering_close()だ。
この処理は、ソケットをクローズする。

   1690             process_timeout_queue(&linger_q, timeout_time, stop_lingering_close);
   1691             /* Step 4: (short) lingering close completion timeouts */

(4)リンガリングクローズタイムアウト(短時間)

short_linger_qキューの処理だ。
ここでのタイムアウト時の処理も、linger_qキューと同じで、stop_lingering_close()だ。
ソケットをクローズする。

   1692             process_timeout_queue(&short_linger_q, timeout_time, stop_lingering_close);
   1693

ここまででタイマーキューの処理は終わりになる。

以下は、いくつかの管理情報の更新処理だ。

   1694             ps = ap_get_scoreboard_process(process_slot);
   1695             ps->write_completion = write_completion_q.count;
   1696             ps->keep_alive = keepalive_q.count;
   1697             apr_thread_mutex_unlock(timeout_mutex);
   1698
   1699             ps->connections = apr_atomic_read32(&connection_count);
   1700             ps->suspended = apr_atomic_read32(&suspended_count);
   1701             ps->lingering_close = apr_atomic_read32(&lingering_count);
   1702         }
   1703         if (listeners_disabled && !workers_were_busy
   1704             && (int)apr_atomic_read32(&connection_count)
   1705                - (int)apr_atomic_read32(&lingering_count)
   1706                < ((int)ap_queue_info_get_idlers(worker_queue_info) - 1)
   1707                  * worker_factor / WORKER_FACTOR_SCALE + threads_per_child)
   1708         {
   1709             listeners_disabled = 0;
   1710             enable_listensocks(process_slot);
   1711         }
   1712         /*
   1713          * XXX: do we need to set some timeout that re-enables the listensocks
   1714          * XXX: in case no other event occurs?
   1715          */
   1716     }     /* listener main loop */

ここまでで主ループは終わる。

listenerスレッドの終了処理では、Listenソケットをクローズし、workerスレッド向けのメッセージキューに終了通知を登録する。

   1717
   1718     close_listeners(process_slot, &closed);
   1719     ap_queue_term(worker_queue);
   1720
   1721     apr_thread_exit(thd, APR_SUCCESS);
   1722     return NULL;
   1723 }

これで、listener_threadの処理を終わる。