Method do_write_request

src/test/msgr/test_async_networkstack.cc:753–793

    }

    void do_write_request() {
      if (dead)
        return ;
      ASSERT_TRUE(socket.is_connected() > 0);

      while (left > 0 && factory->queue_depth > writings.size() + acking.size()) {
        StressFactory::Message *m = new StressFactory::Message(
                factory->rs, ++index,
                factory->rd() % factory->max_message_length);
        std::cerr << " client " << this << " generate message " << m->idx << " length " << m->len << std::endl;
        ASSERT_EQ(m->len, m->content.size());
        writings.push_back(m);
        --left;
        --factory->message_left;
      }

      while (!writings.empty()) {
        StressFactory::Message *m = writings.front();
        bufferlist bl;
        bl.append(m->content.data() + write_offset, m->content.size() - write_offset);
        ssize_t r = socket.send(bl, false);
        if (r == 0)
          break;
        std::cerr << " client " << this << " send " << m->idx << " len " << r << " content: " << std::endl;
        ASSERT_TRUE(r >= 0);
        write_offset += r;
        if (write_offset == m->content.size()) {
          write_offset = 0;
          writings.pop_front();
          acking.push_back(m);
        }
      }
      if (writings.empty() && write_enabled) {
        center->delete_file_event(socket.fd(), EVENT_WRITABLE);
        write_enabled = false;
      } else if (!writings.empty() && !write_enabled) {
        ASSERT_EQ(0, center->create_file_event(
                  socket.fd(), EVENT_WRITABLE, &write_ctxt));
        write_enabled = true;
      }
    }

    bool finish() const {
      return left == 0 && acking.empty() && writings.empty();