| 141 | } |
| 142 | |
| 143 | void RdmaTransport::QueueMessage(InputMessageClosure& input_msg, |
| 144 | int* num_bthread_created, bool last_msg) { |
| 145 | if (last_msg && !rdma::FLAGS_rdma_use_polling) { |
| 146 | return; |
| 147 | } |
| 148 | InputMessageBase* to_run_msg = input_msg.release(); |
| 149 | if (!to_run_msg) { |
| 150 | return; |
| 151 | } |
| 152 | |
| 153 | if (rdma::FLAGS_rdma_disable_bthread) { |
| 154 | ProcessInputMessage(to_run_msg); |
| 155 | return; |
| 156 | } |
| 157 | // Create bthread for last_msg. The bthread is not scheduled |
| 158 | // until bthread_flush() is called (in the worse case). |
| 159 | |
| 160 | // TODO(gejun): Join threads. |
| 161 | bthread_t th; |
| 162 | bthread_attr_t tmp = (FLAGS_usercode_in_pthread ? |
| 163 | BTHREAD_ATTR_PTHREAD : |
| 164 | BTHREAD_ATTR_NORMAL) | BTHREAD_NOSIGNAL; |
| 165 | tmp.keytable_pool = _socket->keytable_pool(); |
| 166 | tmp.tag = bthread_self_tag(); |
| 167 | bthread_attr_set_name(&tmp, "ProcessInputMessage"); |
| 168 | |
| 169 | if (!FLAGS_usercode_in_coroutine && bthread_start_background( |
| 170 | &th, &tmp, ProcessInputMessage, to_run_msg) == 0) { |
| 171 | ++*num_bthread_created; |
| 172 | } else { |
| 173 | ProcessInputMessage(to_run_msg); |
| 174 | } |
| 175 | } |
| 176 | |
| 177 | void RdmaTransport::Debug(std::ostream &os) { |
| 178 | if (_rdma_state == RDMA_ON && _rdma_ep) { |
nothing calls this directly
no test coverage detected