streambuf を拡張し, ソケット対応 [C++]

[2018.6]

C++ の IOストリームで, 新しい対象に対して入出力できるようにしたい場合, istream, ostream, iostream の各クラスに似たストリームクラスと, basic_streambuf クラスから派生させたクラスをペアで作ります。

iostream クラスは書式付き入出力を担当し, 媒体への書き込み・読み込みは streambuf クラスで行なうようになっています。

例えばファイルに対しては fstream クラスと filebuf クラスがペアになります。

ソケットの入出力のストリームクラス, streambuf クラスを作りました。

作ったソースコードはこちら; network · master · netsphere / my-cpp-lib · GitLab

TLS 版については, HTTP/2 のページに掲載しています。HTTP/2 クライアント実装サンプル (TLS版)

仕様

全体的に, なんでこんな仕様になってるの? というところが多い。

書式化は iostream のほうでするのに, ロケールを streambuf のほうに持たせるのは何でだ? streambuf がえらく複雑になっている。

ストリームクラスが streambuf を持つ、というのがチグハグと思う。

バッファリング

読み込みバッファと書き込みバッファを別に設定できる。読み込みバッファは protected setg() で, 書き込みバッファは protected setp() で設定する。(g は get のg, p は put のp. 略さないでほしい...)

これらを呼び出さなければ, バッファリングしない. バッファリングしない場合, public sungetc(), public sputbackc() が失敗する.

読み込み

public sgetn() が固定長の読み込み. 読めた分だけ返すのではなく、指定のバイト数に達するまでブロックする、となっている。::recv() と挙動が違う。recv() のつもりで呼び出すと, スタックしてしまう。

行末まで読みたい、など、どれだけ読めばいいか事前に分からないときは, sgetn() は使えない. そうすると, 1文字ずつ読むしかなく、バッファリングが必須.

実装は, protected xsgetn() に投げてる. そこから uflow() で行っている。uflow() は単に underflow() に投げてる. 

basic_streambuf<>::underflow()
{ return traits_type::eof(); }

このメソッドだけを override すれば, とりあえず動くようになる。

エラーハンドリング

underflow() は, 高々1バイトを読み込む。エラーが起こったときの通知方法がない。握りつぶすよりかは, とりあえず例外でも投げるか。

書き込み側, xsputn() も, エラーが起こったときの通知ができない。こちらも, とりあえず例外にするか。

何でこんなことになっているのか.

ヘッダファイル (実装)

やや長くなるが、全文を示す。

socket_streambuf.h:

C++
[RAW]
  1. #include <ios>
  2. #include <assert.h>
  3. #include <algorithm> // min()
  4. #include <string.h> // memmove()
  5. #ifdef _WIN32
  6. #define STRICT
  7. #define WIN32_LEAN_AND_MEAN
  8. #include <winsock2.h>
  9. #include <windows.h>
  10. #define SHUT_RDWR SD_BOTH
  11. #else
  12. #include <sys/socket.h> // recv(), send(), shutdown()
  13. #include <unistd.h> // close()
  14. typedef int SOCKET;
  15. typedef unsigned char BYTE;
  16. #define INVALID_SOCKET -1
  17. #endif
  20. // TCP socket.
  21. // 非SSL (TLS). TLS 版は TlsSocketStreamBuf クラス.
  22. class SocketStreamBuf: public std::streambuf
  23. {
  24. // sgetn() は読めた分だけ返すのではなく, 指定のバイト数に達するまで,
  25. // ブロックする。
  26. // そのため, どれだけ読むべきか事前に不明な場合は sgetn() は使えず,
  27. // 1バイトずつ読まなければならない。
  28. // 実装としては, バッファが (ほぼ) 必須.
  29. char* m_buf;
  31. #define max_headroom ((ssize_t) 16)
  32. static constexpr long bufsiz = 4096 + max_headroom;
  34. protected:
  35. // 所有.
  36. SOCKET m_sock;
  38. public:
  39. SocketStreamBuf(): m_buf(nullptr), m_sock(INVALID_SOCKET) { }
  41. virtual ~SocketStreamBuf() {
  42. close();
  43. if (m_buf) {
  44. free(m_buf);
  45. m_buf = nullptr;
  46. }
  47. }
  50. // SocketStream::open() から呼び出される.
  51. virtual void connected( SOCKET sock, const char* host ) {
  52. assert( sock != INVALID_SOCKET );
  53. assert( host && *host );
  54. assert( m_sock == INVALID_SOCKET ); // TODO: 例外を投げる?
  56. m_sock = sock;
  57. if (!m_buf)
  58. m_buf = (char*) malloc(bufsiz);
  59. setg(m_buf, m_buf, m_buf);
  60. }
  62. // std::streambuf<> には close() がない. => オブジェクトを破壊する必要.
  63. // @return If failed, -1.
  64. virtual int close() {
  65. if (m_sock == INVALID_SOCKET )
  66. return 0;
  68. // なくてもいいはずだが、OSなどの不具合などによりデータ消失があるた
  69. // め (?), 入れた方がいいという意見もある.
  70. // See TCPのはなし
  71. // http://www.silex.jp/blog/wireless/2015/11/tcp.html
  72. // See close vs shutdown socket?
  73. // https://code.i-harness.com/en/q/3f7b5b
  74. ::shutdown(m_sock, SHUT_RDWR);
  76. int r;
  77. #ifdef _WIN32
  78. r = ::closesocket(m_sock);
  79. #else
  80. r = ::close(m_sock);
  81. #endif
  82. m_sock = INVALID_SOCKET;
  83. return r;
  84. }
  87. bool is_open() const throw() {
  88. return m_sock != INVALID_SOCKET;
  89. }
  92. protected:
  93. // バッファを無視して, 読み込む.
  94. // length に満たない場合がある.
  95. // @return エラーの場合, -1
  96. virtual ssize_t sysread(void* buffer, size_t length) {
  97. ssize_t ret;
  98. do {
  99. ret = ::recv(m_sock, (char*) buffer, length, 0);
  100. } while (ret == -1 && errno == EINTR);
  101. return ret;
  102. }
  104. // 読み込み用で最低限 override しなければならないのは, underflow() だけ.
  105. // @return 読み込んだ 1文字. EOF だった場合, traits_type::eof().
  106. virtual int_type underflow() {
  107. if ( !is_open() )
  108. return traits_type::eof();
  110. // protected eback() 先頭, gptr() 現在の位置, egptr() 終端. 名前が??
  111. if (!gptr() || gptr() >= egptr()) {
  112. // バッファを読み切った.
  113. // sungetc() できるようにする.
  114. long rest = std::min(egptr() - eback(), max_headroom);
  115. if (rest > 0)
  116. memmove(m_buf, gptr() - rest, rest);
  117. ssize_t r = this->sysread(m_buf + rest, bufsiz - rest);
  118. if (r == 0) // EOF
  119. return traits_type::eof();
  120. else if (r < 0)
  121. throw errno;
  122. setg(m_buf, m_buf + rest, m_buf + rest + r);
  123. }
  125. return *gptr(); // ポインタ進めない.
  126. }

上述のとおり underflow() だけを override すればとりあえず動くが、バッファリングして, sysread() でドカンと読み込むようにする。

C++
[RAW]
  1. // 同様に, 書き込み側については, 最低限 overflow() を override すればよい.
  2. // 1バイトを書き込む.
  3. // @param __c 書き込む値.
  4. // @return 成功したら EOF 以外の何か. 失敗したら EOF.
  5. // ここの解説では, __c が EOF の場合, 失敗 (EOF) を返す.
  6. // http://www.cplusplus.com/reference/streambuf/basic_streambuf/overflow/
  7. // basic_filebuf<>::overflow() の実装では, __c が EOF の場合も
  8. // 「成功」を返している.
  9. virtual int_type overflow(int_type __c = traits_type::eof() )
  10. {
  11. if ( traits_type::eq_int_type(__c, traits_type::eof()) )
  12. return __c; // error とする.
  13. char cc = __c;
  14. if ( xsputn(&cc, 1) <= 0 )
  15. return traits_type::eof();
  17. return 1; // 成功
  18. }
  20. // 効率悪いので, xsputn() も override.
  21. // @return 書き込めたバイト数.
  22. // basic_streambuf<>::xsputn() の実装を見ると, overflow() が EOF
  23. // を返すと, そこまでのバイト数を返す.
  24. // エラーで1バイトも書き込めなかったときも, 0 を返す. マジか!!
  25. virtual std::streamsize xsputn( const char_type* s,
  26. std::streamsize count )
  27. {
  28. if ( !is_open() )
  29. return 0;
  30. ssize_t r = ::send(m_sock, s, count, 0 );
  31. if (r >= 0)
  32. return r;
  33. else
  34. throw errno;
  35. }
  36. };