大路・小路　ゆっくり、ほっこり

今回は受信側(サーバ)のプログラムを考えてみましょう。
CSendRecvThraedの受信リングバッファに格納されたファイル送信コマンドを
CRecvFileThreadのファイル受信用のリングバッファに格納します。
CRecvFileThreadはこのリングバッファに格納されたファイル送信コマンドに
従って処理を行います。
今回は簡単のためにファイル受信処理でエラーが発生した時は、CRecvFileThreadを
終了、CSendRecvThraedも終了して当該クライアントとの接続を終了(切断)する
ことにします。

ファイル受信を実現するクラスCRecvFileThreadの作成
次のような機能を考えます。
CSendRecvThreadでいつでもファイル受信ができるようにCSendRecvThread構築時に
CRecvFileThreadを構築します。
破棄については、CSendRecvThreadが破棄時されるときに破棄します。
AnalyzeCmd()でm_pCRingBuffCmdに格納されているコマンドを解析し以下を実行します。
(＊)コマンドが実行状態(遷移状態)と一致しないときはエラーとして切断します。
　エラー発生時、CRecvFileThreadをゾンビ状態にして、CSendRecvThreadで破棄し
　CSendRecvThreadをゾンビ状態にすると、当該のクライアントは切断されます。
　切断せずにエラーをクライアントに送信し新たなファイルを送信するようにもできますが、
　簡単のために今回は切断することにします。
①指定されたファイルをオープンします。
　RecvSendFileStartPacket(SendFileStartReqRec *pbPacket)
(＊)結果を送信する
②ファイルを書き込みます。
　 RecvSendFileIngPacket(SendFileIngRec *pbPacket)
②を繰り返します。
③ファイルをクローズします。
　RecvSendFileEndPacket(SendFileEndReqRec *pbPacket)
(＊)結果を送信します。
③まで、正常に完了すれば、次のファイルを受信することができます。
(＊)については、後程実装することにします。

プロジェクト(SimpleServerTransFile)に次の変更を行います。
今回の仕様では、ファイルの置き場所をカレントフォルダにするので
デバッグ時も実行ファイルがある場所をカレントフォルダとするように
プロパティの変更を行います。
作業ディレクトリの変更
　/home/pi/projects/$(ProjectName)/に変更します。

Windowsとの互換性を図るためにSimplServer.cppに記述して以下の内容を
stdThread.hに移動します。

#ifndef _stdThread_H_2B9D23FC_90BF_4898_B161_B894F01F6698 #define _stdThread_H_2B9D23FC_90BF_4898_B161_B894F01F6698 #include <stdio.h> #include <stdlib.h> #include <pthread.h> #include <unistd.h> #include <string.h> #include <sys/time.h> #include <sys/select.h> #include <netdb.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <errno.h> #include <netinet/tcp.h> #include <sys/stat.h> #include <sys/types.h> #include <fcntl.h> #include <limits.h> #include <termios.h> #include <poll.h> #define SOCKET_ERROR (-1) #define INVALID_SOCKET (-1) #define TRUE (1) #define FALSE (0) typedef unsigned char BYTE; typedef int BOOL; typedef unsigned int UINT; typedef unsigned short WORD; // Linux :LP64 long:64 int:32 // Windows:LLP64 long:32 int:32 pointerのみが64bitsになっている //typedef unsigned long DWORD; // Windows typedef unsigned int DWORD; // Linux typedef BYTE *LPBYTE; typedef char *LPSTR; typedef const char *LPCSTR; typedef char *LPTSTR; typedef const char *LPCTSTR; typedef signed long long SInt64; typedef unsigned long long UInt64; typedef int SOCKET; //======================================================================= #define min(a,b) (((a) < (b)) ? (a) : (b)) #define max(a,b) (((a) > (b)) ? (a) : (b)) #define SAFE_FREE(p) { if(p) { free(p); (p)=NULL; } } #define SAFE_DELETE(p) { if(p) { delete (p); (p)=NULL; } } #define SAFE_DELETE_ARRAY(p) { if(p) { delete[] (p); (p)=NULL; } } //======================================================================= // ★ #ifdef _MSC_VER // Windowsのとき #define DISABLE_C4996 __pragma(warning(push)) __pragma(warning(disable:4996)) #define ENABLE_C4996 __pragma(warning(pop)) #else // Linuxのとき #define DISABLE_C4996 #define ENABLE_C4996 #endif extern DWORD GetdwInterval(DWORD dwNow, DWORD dwPast); extern DWORD timeGetTime(); extern int kbhit(void); extern int DispErrorMsg(LPCSTR pszTitle); extern void Locate(int iCol, int iRow, int iErase); #endif

プロジェクトにクラスCRecvThread(RecvFileThread.h, RecvFileThread.cpp)を追加します。
基底クラスはCThreadJobです。

RecvFileThread.hは以下のようになります。
コンストラクタでは送受信スレッド(親スレッド)を渡します。
①指定されたファイルをオープンする、②ファイルを書き込む、③ファイルをクローズする
を実現するメソッドを定義しています。
またエラー時にファイルを削除するためにファイル名を記憶する変数を定義します。

【RecvFileThread.h】 #pragma once #include “ThreadJob.h” #include “define.h” // パケット作成のため #define CMD_BUFF_SIZE (1024 * 64) // ファイル送信コマンド格納リングバッファのサイズ // ファイル受信遷移状態 #define RECVC_FILE_STAT_IDLE 0 // 初期状態 #define RECVC_FILE_STAT_ING 1 // ファイル受信中 class CMySyncObject; // CMySyncObjectを使うため class CSendRecvThread; // CSendRecvThreadを使うため class CRingBuff; // CRingBuffを使うため class CRecvFileThread : public CThreadJob { public: CRecvFileThread(CSendRecvThread *pCSendRecvThread); ~CRecvFileThread(); CRingBuff *m_pCRingBuffCmd; // ファイル送信コマンド格納リングバッファ UINT DoWork() override; BOOL IsZombie(); // このスレッドはゾンビ状態か private: CSendRecvThread *m_pCSendRecvThread; int m_fdDstFile; // 受信ファイルディスクリプタ CMySyncObject *m_pCMySyncObject; BOOL m_fIamZombie; // ゾンビ状態か int m_iRecvFileStatus; // 遷移状態 RECVC_FILE_STAT_IDLE/RECVC_FILE_STAT_ING // ファイル名文字列(nullターミネートを含む)UTF-8 エラー時ファイル削除用 char m_szFileName[FILE_NAME_LEN + 1]; int AnalyzeCmd(); int RecvSendFileStartPacket(SendFileStartReqRec *pbPacket); // ①指定されたファイルをオープンする int RecvSendFileIngPacket(SendFileIngRec *pbPacket); // ②ファイルを書き込む int RecvSendFileEndPacket(SendFileEndReqRec *pbPacket); // ③ファイルをクローズする };

RecvFileThread.cppは以下のようになります。
ファイルを書き込みモードでオープンしますが、すでに存在しているファイルは
長さを0にしてしまうことにします。(一時ファイルを作成し、ファイル受信が成功したら
名前を変更する方が良いのですが、ここでは簡単のために同名ファイルは上書きする
ことにしています）
　m_fdDstFile = open(pbPacket->FileInfo.szFileName,
　　O_CREAT | O_WRONLY | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP |
　　S_IROTH | S_IWOTH);
　のようにLinuxではモードも指定します。
　引数の詳細は次のサイトを参照してください。
　Linux Programmer’s Manual

【RecvFileThread.cpp】 #include “RecvFileThread.h” #include “MySyncObject.h” #include “SendRecvThread.h” #include “RingBuff.h” CRecvFileThread::CRecvFileThread(CSendRecvThread *pCSendRecvThread) { // パラメータを記憶する m_pCSendRecvThread = pCSendRecvThread; // 変数の初期化 m_fIamZombie = FALSE; m_fdDstFile = -1; // ファイルディスクリプタ m_iRecvFileStatus = RECVC_FILE_STAT_IDLE; // ファイル送信コマンド処理の遷移 memset(m_szFileName, 0, sizeof(m_szFileName)); // ファイル削除のためのファイル名の初期化 m_pCRingBuffCmd = new CRingBuff(CMD_BUFF_SIZE); // ファイル送信コマンド格納用 m_pCMySyncObject = new CMySyncObject(); m_pCMySyncObject->Initialize(); } CRecvFileThread::~CRecvFileThread() { if (m_pCMySyncObject != NULL) m_pCMySyncObject->Uninitialize(); SAFE_DELETE(m_pCMySyncObject) SAFE_DELETE(m_pCRingBuffCmd) // オープンしたままのファイルがあればおかしいのでクローズ、削除 if (m_fdDstFile != -1) { close(m_fdDstFile); unlink(m_szFileName); } m_fdDstFile = -1; } UINT CRecvFileThread::DoWork() { BOOL fRet = TRUE; while (!m_fStopFlag) { usleep(10 * 1000); // usleepを入れて他のスレッドにチャンスを与える if (AnalyzeCmd() == -1) // 返値が-1の時がエラー、このスレッドをゾンビにする { fRet = FALSE; break; } } m_pCMySyncObject->Lock(); m_fIamZombie = TRUE; m_pCMySyncObject->UnLock(); return((fRet == TRUE) ? 0 : -1); } //============================================== // function // ファイル送信コマンドの解析 // parameter // なし // retun // 0:十分なデータが格納されていないなど 1:解析成功 -1:解析失敗 //============================================== int CRecvFileThread::AnalyzeCmd() { int iRet = 0; HeaderRec Header; WORD wCmd; int iSize = m_pCRingBuffCmd->GetReadableSize(); int iPacketSize, iDataSize; LPBYTE pbPacket = NULL; if (iSize < sizeof(HeaderRec)) // ヘッダサイズに満たないときは何もしない goto L_END; // ヘッダ部を借り読み込みする(パケット丸ごと格納されているはずですが) m_pCRingBuffCmd->ReadWithoutUpdateHeadPoint((LPBYTE)&Header, sizeof(HeaderRec)); wCmd = ntohs(Header.wCommand); iDataSize = ntohs(Header.wDataLen); iPacketSize = iDataSize + sizeof(HeaderRec); if (iSize < iPacketSize) // パケットサイズに満たないときは何もしない goto L_END; pbPacket = (LPBYTE)calloc(iPacketSize, sizeof(BYTE)); // パケット全体を読む m_pCRingBuffCmd->Read(pbPacket, iPacketSize); switch (wCmd) { case CMD_SEND_FILE_START_REQ: // ①送信開始要求受信 iRet = RecvSendFileStartPacket((SendFileStartReqRec *)pbPacket); break; case CMD_SEND_FILE_ING: // ②ファイル送信中 iRet = RecvSendFileIngPacket((SendFileIngRec *)pbPacket); break; case CMD_SEND_FILE_END_REQ: // ③送信完了要求受信 iRet = RecvSendFileEndPacket((SendFileEndReqRec *)pbPacket); break; // ★CMD_SEND_FILE_ABORT_REQも後ほど処理するようにしましょう default: iRet = -1; // 知らないコマンド break; } L_END: SAFE_FREE(pbPacket) return(iRet); } //============================================== // function // ①ファイル送信開始コマンドの受信 // parameter // SendFileStartReqRec *pbPacket [in]ファイル送信開始コマンドパケット // retun // 1:解析成功 -1:致命的エラー //============================================== int CRecvFileThread::RecvSendFileStartPacket(SendFileStartReqRec *pbPacket) { int iRet = 0; Locate(1, 11, 2); fprintf(stderr, “Start %s %s(%u)”, pbPacket->FileInfo.szFileName, pbPacket->FileInfo.szFileSize, timeGetTime()); if (m_iRecvFileStatus != RECVC_FILE_STAT_IDLE) { iRet = -1; // 遷移状態エラーは切断する goto L_END; } // ここで書き込み用にファイルをopenする if ((m_fdDstFile = open(pbPacket->FileInfo.szFileName, O_CREAT | O_WRONLY | O_TRUNC, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH)) == -1) { // ★後程ファイルが作れないエラーをクライアント返すようにします goto L_END; } // 開始したファイル名を覚える strcpy(m_szFileName, pbPacket->FileInfo.szFileName); m_iRecvFileStatus = RECVC_FILE_STAT_ING; // ★後程ファイル送信開始要求成功を返すようにします iRet = 1; L_END: if (iRet == -1) // エラー時オープンしているファイルがあればクローズ、削除 { if (m_fdDstFile != -1) { close(m_fdDstFile); unlink(m_szFileName); } m_fdDstFile = -1; } return(iRet); } //============================================== // function // ファイル送信中コマンドの受信 // parameter // SendFileIngRec *pbPacket[in]ファイル送信中コマンドパケット // return // 0:十分なデータが格納されていない 1:解析成功 -1:解析失敗 //============================================== int CRecvFileThread::RecvSendFileIngPacket(SendFileIngRec *pbPacket) { int iRet = 0; long lDataSize, lOffset; Locate(1, 12, 1); fprintf(stderr, “Ing %s %s”, pbPacket->Stat.szOffset, pbPacket->Stat.szSize); if (m_iRecvFileStatus != RECVC_FILE_STAT_ING) { iRet = -1; // 遷移状態エラーは切断する goto L_END; } lDataSize = atol(pbPacket->Stat.szSize); lOffset = atol(pbPacket->Stat.szOffset); if (lDataSize > 0) { // ファイル操作失敗時エラーとして切断 if (lseek(m_fdDstFile, lOffset, SEEK_SET) == -1) { iRet = -1; goto L_END; } if (write(m_fdDstFile, pbPacket->bData, lDataSize) != lDataSize) { iRet = -1; goto L_END; } } iRet = 1; L_END: if (iRet == -1) // エラー時オープンしているファイルがあればクローズ、削除 { if (m_fdDstFile != -1) { close(m_fdDstFile); unlink(m_szFileName); } m_fdDstFile = -1; } return(iRet); } //============================================== // function // ファイル送信終了コマンドの受信 // parameter // SendFileEndReqRec *pbPacket [in]ファイル送信終了コマンドパケット // return // 0:十分なデータが格納されていない 1:解析成功 -1:解析失敗 //============================================== int CRecvFileThread::RecvSendFileEndPacket(SendFileEndReqRec *pbPacket) { int iRet = 0; long lDataSize, lOffset; Locate(1, 13, 1); fprintf(stderr, “End %s %s(%u)”, pbPacket->Stat.szOffset, pbPacket->Stat.szSize, timeGetTime()); if (m_iRecvFileStatus != RECVC_FILE_STAT_ING) { iRet = -1; // 遷移状態エラーは切断する goto L_END; } lDataSize = atol(pbPacket->Stat.szSize); lOffset = atol(pbPacket->Stat.szOffset); if (lDataSize > 0) { // ファイル操作失敗時エラーとして切断 if (lseek(m_fdDstFile, lOffset, SEEK_SET) == -1) { iRet = -1; goto L_END; } if (write(m_fdDstFile, pbPacket->bData, lDataSize) != lDataSize) { iRet = -1; goto L_END; } } iRet = 1; // ★後程送信完了要求成功を返すようにします L_END: if (m_fdDstFile != -1) { close(m_fdDstFile); if(iRet == -1) // エラー時削除 unlink(m_szFileName); } m_iRecvFileStatus = RECVC_FILE_STAT_IDLE; m_fdDstFile = -1; return(iRet); } //============================================== // function // このスレッドはゾンビ状態か // 別スレッドから参照される // parameter // なし // return // 0:正常 -1:エラー発生 //============================================== BOOL CRecvFileThread::IsZombie() { BOOL fRet; m_pCMySyncObject->Lock(); fRet = m_fIamZombie; m_pCMySyncObject->UnLock(); return(fRet); }

CSendRecvThreadにCRecvFileThreadの呼び出し(構築・破棄)を追加します。
いつでもファイル受信ができるように、コンストラクタでCRecvFileThreadを構築し
開始します。
CRecvFileThreadでエラーが発生(Zombie状態)していたらCRecvFileThreadを破棄し
自分自身(CSendRecvThread)をZombieにし、メインスレッドから切断をさせる
ようにします。
先ずは、SendRecvThread.hです。変更点には★を付けました。

【SendRecvThread.h】 #pragma once #include “ThreadJob.h” #include “define.h” // RecvMessagePacketの引数のため class CMySyncObject; // このクラスの使用することを宣言 class CRingBuff; // CRingBuffを使用するため class CRecvFileThread; // ★CRecvFileThreadを使用するため #define SEND_BUFF_SIZE (1024 * 64) // 送信リングバッファのサイズ #if 1 #define SENDBUFSIZE (1024 + sizeof(HeaderRec)) // 一度に送信するサイズの最大値(共通ヘッダ分を加えてあります) #else #define SENDBUFSIZE (1024 * 4) // TCPで送信速度を1Mbpsより上げたいときは断片化処理をTCPのスタックに任せる // スタック(send)に渡すサイズを増やす #endif #define RECV_BUFF_SIZE (1024 * 64) // 受信リングバッファのサイズ #define RCVBUFSIZE (1024 * 2) // 一度に読む最大受信サイズ typedef struct { SOCKET fdClient; // 接続済みソケット(acceptの結果) CMySyncObject *pCMySyncObject; // 同期オブジェクト } ConnectionInfoRec; class CSendRecvThread : public CThreadJob { public: CSendRecvThread(ConnectionInfoRec *pConInfo); // パラメータをコンストラクタで渡す ~CSendRecvThread(); // 基底クラスの関数をオーバーライドする // C++11で明示的にoverrideを書くことが出来るようになりました // 基底クラスの当該関数にvirtualが書いていないとエラーを出してくれます UINT DoWork() override; // DoRecvで実施している内容を記述 BOOL SetSendData(char *pcData, int iSize); // 送信データの設定 int GetDataForDistributeChat(char **ppcData); // チャット分配用データの取得 BOOL IsZombie(); // このスレッドはゾンビ状態か private: ConnectionInfoRec *m_pConInfo; // コンストラクタで渡されるパラメータを格納 // このスレッド実行中領域が確保されていること BOOL m_fIamZombie; // ゾンビ状態かどうかを保持 CRingBuff *m_pCRingBuffSend; // 送信データ格納用リングバッファ CRingBuff *m_pCRingBuffRecv; // 受信データ格納用リングバッファ CRingBuff *m_pCRingBuffForDistributeChat; // チャットメッセージをクライアントに配信するためのリングバッファ DWORD m_dwPrevSentTime; // 前回送信した時刻(msec) DWORD m_dwSendInterval; // 次回送信までの間隔(msec) CRecvFileThread *m_pCRecvFileThread; // ★ファイル受信スレッド int GetSendData(char **ppcData); // 送信データの取得 BOOL SetDataForDistributeChat(char *pcData, int iSize); // チャットメッセージをクライアントに分配するために格納する int AnalyzeDataRecv(); int RecvMessagePacket(HeaderRec *pHeader); int RecvFileSendPacket(HeaderRec *pHeader); // ★ファイル送信コマンド受信関数 BOOL SendMessagePacket(char *pcData, int iSize); DWORD CalcNextSendInterval(int iSentSize); // 次回送信までの間隔 BOOL CanSendNow(DWORD dwNow); // 送信して良い時刻になったか BOOL KillZombei(); // ★このクラスが作った不要(エラー発生)になったスレッドを破棄 };

次に、SendRecvThread.cppを見てみましょう。
変更点には★を付けました。
変更はコンストラクタ、デストラクタと次のメソッドです。
UINT CSendRecvThread::DoWork()
BOOL CSendRecvThread::AnalyzeDataRecv()
int CSendRecvThread::RecvFileSendPacket(HeaderRec *pHeader)
BOOL CSendRecvThread::KillZombei()

【SendRecvThread.cpp】 #include “SendRecvThread.h” #include “MySyncObject.h” // CMySyncObjectを使うため #include “RingBuff.h” // CRingBuffを使うため #include “RecvFileThread.h” // ★CRecvFileThreadを使うため // 送信速度 #define SEND_BPS (1000000.0) // 送信速度1Mbps //============================================== // function // コンストラクタ // parameter // ConnectionInfoRec *pConInfo [in]機能に必要な情報 // return // なし //============================================== CSendRecvThread::CSendRecvThread(ConnectionInfoRec *pConInfo) { m_pConInfo = pConInfo; m_fIamZombie = FALSE; m_pCRingBuffSend = new CRingBuff(SEND_BUFF_SIZE); // 送信リングバッファの構築 m_pCRingBuffRecv = new CRingBuff(RECV_BUFF_SIZE); // 受信リングバッファの構築 m_pCRingBuffForDistributeChat = new CRingBuff(RECV_BUFF_SIZE); // チャット配信用リングバッファの構築 m_dwPrevSentTime = 0; // 初回送信はすぐに m_dwSendInterval = 0; // m_pCRecvFileThread = new CRecvFileThread(this); // ★ファイル受信スレッドを常に使えるように準備する m_pCRecvFileThread->Begin(); } //============================================== // function // デストラクタ // parameter // なし // return // なし //============================================== CSendRecvThread::~CSendRecvThread() { if (m_pCRecvFileThread != NULL) // ★このクラスが破棄されるときにファイル受信スレッドを破棄 { m_pCRecvFileThread->End(); m_pCRecvFileThread->WaitForEnd(); } SAFE_DELETE(m_pCRecvFileThread) SAFE_DELETE(m_pCRingBuffSend) // 送信リングバッファの破棄 SAFE_DELETE(m_pCRingBuffRecv) // 受信リングバッファの破棄 SAFE_DELETE(m_pCRingBuffForDistributeChat) // チャット配信用リングバッファの破棄 } //============================================== // function // 機能を記述した関数 // parameter // なし // return // 0:正常 -1:エラー発生 //============================================== UINT CSendRecvThread::DoWork() { BOOL fRet = TRUE; char szRecvBuffer[RCVBUFSIZE]; // 受信バッファ int iRecvSize; pollfd fds[1] = { 0 }; char *pcData = NULL; // 未送信データ int iSendSize = 0; // 未送信データサイズ DWORD dwNow; // 送信チェックした時刻を覚えるため fprintf(stderr, “DoWork()\n”); fds[0].fd = m_pConInfo->fdClient; while (!m_fStopFlag) { // ★致命的エラー発生のスレッドを破棄したときは終了 // 終了すると当該クライアントは切断される if (KillZombei() == TRUE) { fRet = FALSE; break; } // 送信データがあるかリングバッファを調べる iSendSize = m_pCRingBuffSend->GetReadableSize(); fds[0].events = POLLIN | POLLRDHUP; // 受信と相手側からの切断イベントを設定 // チェックのために現在時刻を取得する dwNow = timeGetTime(); // 送信可能時刻かつ未送信データがあるときだけ送信可能検査 if ((iSendSize > 0) && (CanSendNow(dwNow) == TRUE)) fds[0].events |= POLLOUT; // 書き込み可能を追加 poll(fds, 1, 10); if (fds[0].revents & POLLRDHUP) // 相手側からの切断 { fprintf(stderr, “Disconnected pollrdhup\n”); fRet = FALSE; break; } if (fds[0].revents & POLLERR) // エラー発生 { m_pConInfo->pCMySyncObject->Lock(); DispErrorMsg(“Err:DoWork”); m_pConInfo->pCMySyncObject->UnLock(); fRet = FALSE; break; } if (fds[0].revents & POLLIN) // 受信データイベント { // 受信リングバッファに空きがあれば受信する // ここで受信しなければ、次回のrevents検査で受信が行われる iRecvSize = min(m_pCRingBuffRecv->GetWriteableSize(), RCVBUFSIZE); if (iRecvSize > 0) { if ((iRecvSize = recv(m_pConInfo->fdClient, szRecvBuffer, iRecvSize, 0)) <= 0) { m_pConInfo->pCMySyncObject->Lock(); if (iRecvSize == 0) DispErrorMsg(“Disconnected recv”); else DispErrorMsg(“Err:recv”); m_pConInfo->pCMySyncObject->UnLock(); fRet = FALSE; break; } else { // 取得したデータすぐに受信リングバッファに書きこむ // 受信リングバッファに書き込むのはこのスレッドだけなので、すべて書き込めるはず m_pCRingBuffRecv->Write((LPBYTE)szRecvBuffer, iRecvSize); } } } // 受信リングバッファに格納されているデータの解析を行う // 複数のパケットが格納されている可能性があるので、reventsの結果とは無関係に // 解析を行うようにする if (AnalyzeDataRecv() == -1) { m_pConInfo->pCMySyncObject->Lock(); DispErrorMsg(“Err:Packet format”); m_pConInfo->pCMySyncObject->UnLock(); fRet = FALSE; break; } if (fds[0].revents & POLLOUT) // 送信可能ならsend実施 { // 送信したいデータの取得(PATH_MTUより小さくなるように取得する) iSendSize = GetSendData(&pcData); if (send(m_pConInfo->fdClient, pcData, iSendSize, 0) != iSendSize) { DispErrorMsg(“Err:send”); fRet = FALSE; break; } SAFE_FREE(pcData) // 送信時刻と次回送信までの時間をセット m_dwPrevSentTime = dwNow; m_dwSendInterval = CalcNextSendInterval(iSendSize); } } m_pConInfo->pCMySyncObject->Lock(); m_fIamZombie = TRUE; m_pConInfo->pCMySyncObject->UnLock(); SAFE_FREE(pcData) return((fRet == TRUE) ? 0 : -1); } //============================================== // function // 送信データの設定 // 送信できない時、エラーとして切断をするのが良いか // は迷うところ // parameter // char *pcData [in]送信データ // int iSize [in]データ長 // return // TRUE/FALSE //============================================== BOOL CSendRecvThread::SetSendData(char *pcData, int iSize) { BOOL fRet = FALSE; // 送信リングバッファに空きがないときは書き込まない fRet = m_pCRingBuffSend->Write((LPBYTE)pcData, iSize); return(fRet); } //============================================== // function // 送信データの取得 // parameter // char **ppcData [in/out]送信データ // return // データ長 //============================================== int CSendRecvThread::GetSendData(char **ppcData) { int iSize = 0; // 送信データがあるかリングバッファのデータサイズを調べる if ((iSize = m_pCRingBuffSend->GetReadableSize()) > 0) { // 送信サイズをPATH_MTUより小さくしておく(1024+共通ヘッダなら大丈夫) iSize = min(iSize, SENDBUFSIZE); *ppcData = (char *)calloc(iSize, sizeof(char)); iSize = m_pCRingBuffSend->Read((LPBYTE)*ppcData, iSize); } return(iSize); } //============================================== // function // チャットメッセージをクライアントに分配するために格納する // parameter // char *pcData [in]チャットメッセージパケット // int iSize [in]パケット長 // return // TRUE/FALSE //============================================== BOOL CSendRecvThread::SetDataForDistributeChat(char *pcData, int iSize) { BOOL fRet = FALSE; // 格納用リングバッファに空きがないときは書き込まない fRet = m_pCRingBuffForDistributeChat->Write((LPBYTE)pcData, iSize); return(fRet); } //============================================== // function // チャット分配用データの取得 // パケット単位で取り出す // 今回は１パケット分 // parameter // char **ppcData [in/out]チャット転送用データ // return // データ長 //============================================== int CSendRecvThread::GetDataForDistributeChat(char **ppcData) { int iSize = 0, iStoreSize, iPacketSize; HeaderRec Header; if ((iStoreSize = m_pCRingBuffForDistributeChat->GetReadableSize()) >= sizeof(HeaderRec)) { // ヘッダ部を仮り読み込みする m_pCRingBuffForDistributeChat->ReadWithoutUpdateHeadPoint((LPBYTE)&Header, sizeof(HeaderRec)); iPacketSize = ntohs(Header.wDataLen) + sizeof(HeaderRec); if (iStoreSize >= iPacketSize) { *ppcData = (char *)calloc(iPacketSize, sizeof(char)); iSize = m_pCRingBuffForDistributeChat->Read((LPBYTE)*ppcData, iPacketSize); } } return(iSize); } //============================================== // function // このスレッドはゾンビ状態か // 別スレッドから参照される // parameter // なし // return // 0:正常 -1:エラー発生 //============================================== BOOL CSendRecvThread::IsZombie() { BOOL fRet; m_pConInfo->pCMySyncObject->Lock(); fRet = m_fIamZombie; m_pConInfo->pCMySyncObject->UnLock(); return(fRet); } //============================================== // function // 受信データの解析 // 受信リングバッファに格納されているデータを調べる // parameter // なし // return // 0:パケットが揃っていない // 1:パケットを受信したので処理を行った // -1:エラーが発生した //============================================== BOOL CSendRecvThread::AnalyzeDataRecv() { int iRet = 0; HeaderRec Header; WORD wCmd; // データサイズを調べる int iSize = m_pCRingBuffRecv->GetReadableSize(); if (iSize < sizeof(HeaderRec)) // ヘッダサイズに満たないときは何もしない goto L_END; // ヘッダ部を借り読み込みする m_pCRingBuffRecv->ReadWithoutUpdateHeadPoint((LPBYTE)&Header, sizeof(HeaderRec)); // ヘッダ部の解析 if (memcmp(Header.bMagicData, MAGIC_STRING, strlen(MAGIC_STRING)) != 0) { iRet = -1; // 識別子が違うのでエラー goto L_END; } wCmd = ntohs(Header.wCommand); Locate(1, 10, 1); fprintf(stderr, “CMD:%d”, wCmd); switch (wCmd) { case CMD_MSG_DATA: iRet = RecvMessagePacket(&Header); break; // ★ファイル送信コマンド(C->S)を処理する case CMD_SEND_FILE_START_REQ: case CMD_SEND_FILE_ING: case CMD_SEND_FILE_END_REQ: case CMD_SEND_FILE_ABORT_REQ: iRet = RecvFileSendPacket(&Header); break; default: // 知らないコマンド iRet = -1; break; } L_END: return(iRet); } //============================================== // function // メッセージコマンドの受信 // parameter // HeaderRec *pHeader [in]仮読みしたヘッダ // return // 0:パケットが揃っていない // 1:パケットを受信したので処理を行った // -1:エラーが発生した //============================================== BOOL CSendRecvThread::RecvMessagePacket(HeaderRec *pHeader) { int iRet = 0; MsgDataRec *pMsgData; int iMsgSize, iSize; char *pszMsg = NULL; iMsgSize = ntohs(pHeader->wDataLen); // データが足りないときは何もしない iSize = m_pCRingBuffRecv->GetReadableSize(); if (iSize < iMsgSize + sizeof(HeaderRec)) goto L_END; // パケット全体を受信しているので読み込みを実施する pMsgData = (MsgDataRec *)calloc(iMsgSize + sizeof(HeaderRec), sizeof(BYTE)); m_pCRingBuffRecv->Read((LPBYTE)pMsgData, iMsgSize + sizeof(HeaderRec)); // NULLターミネート分を追加して確保 pszMsg = (char *)calloc(iMsgSize + 1, sizeof(char)); memcpy(pszMsg, pMsgData->bMsgData, iMsgSize); // チャット送信のためにメッセージデータをパケット化してセット SendMessagePacket(pszMsg, iMsgSize); m_pConInfo->pCMySyncObject->Lock(); fprintf(stderr, “Msg:recv %s\n”, pszMsg); m_pConInfo->pCMySyncObject->UnLock(); SAFE_FREE(pMsgData) SAFE_FREE(pszMsg) iRet = 1; L_END: return(iRet); } //============================================== // function // ★ファイル送信コマンドの受信 // ファイル送信コマンド(C->S)をCRecvThreadに渡すだけ // parameter // HeaderRec *pHeader [in]仮読みしたヘッダ // return // 0:パケットが揃っていない, ファイル受信側が満杯 // 1:パケットを受信したので処理を行った // -1:エラーが発生した //============================================== int CSendRecvThread::RecvFileSendPacket(HeaderRec *pHeader) { int iRet = 0; int iDataSize, iSize, iPacketSize; LPBYTE pbPacket = NULL; iDataSize = ntohs(pHeader->wDataLen); iPacketSize = iDataSize + sizeof(HeaderRec); iSize = m_pCRingBuffRecv->GetReadableSize(); if (iSize < iPacketSize) goto L_END; pbPacket = (LPBYTE)calloc(iPacketSize, sizeof(BYTE)); // パケット全体を仮読み m_pCRingBuffRecv->ReadWithoutUpdateHeadPoint(pbPacket, iPacketSize); if (m_pCRecvFileThread != NULL) { // ファイル受信スレッドに渡せたら読み込んだことにする if (m_pCRecvFileThread->m_pCRingBuffCmd->Write(pbPacket, iPacketSize) == TRUE) { m_pCRingBuffRecv->UpdateHeadPoint(iPacketSize); iRet = 1; } } else { // ファイル受信スレッドがいなければエラー切断 m_pCRingBuffRecv->UpdateHeadPoint(iPacketSize); iRet = -1; } SAFE_FREE(pbPacket); L_END: return(iRet); } //============================================== // function // メッセージパケットの送信 // チャットメッセージをクライアントに転送するためのエリアに // 格納する // parameter // char *pcData [in]UTF-8データ(NULLターミネートなし) // int iSize [in]データサイズ // retun // なし //============================================== BOOL CSendRecvThread::SendMessagePacket(char *pcData, int iSize) { MsgDataRec *pMsgData = NULL; int iPacketSize = sizeof(HeaderRec) + iSize; // パケットサイズ BOOL fRet = FALSE; pMsgData = (MsgDataRec *)calloc(iPacketSize, sizeof(BYTE)); // パケット全体のエリアを確保 memcpy(pMsgData->header.bMagicData, MAGIC_STRING, strlen(MAGIC_STRING)); pMsgData->header.wCommand = htons(CMD_MSG_DATA); pMsgData->header.wDataLen = htons(iSize); memcpy(pMsgData->bMsgData, pcData, iSize); // チャット送信のためにメッセージパケットデータをセット fRet = SetDataForDistributeChat((char *)pMsgData, iPacketSize); SAFE_FREE(pMsgData) return(fRet); } //============================================== // function // 次の送信までの間隔(msec)を求める // parameter // int iSentSize [in]送信したサイズ // return // msec //============================================== DWORD CSendRecvThread::CalcNextSendInterval(int iSentSize) { DWORD dwInterval = 0; double dbBytePerSec = SEND_BPS / 8.0; double dbCountPerSec; if (iSentSize == 0) goto L_END; dbCountPerSec = dbBytePerSec / (double)iSentSize; dwInterval = DWORD(1000.0 / dbCountPerSec); L_END: return(dwInterval); } //============================================== // function // 今送信可能時刻か // parameter // DWORD dwNow [in]現在時刻 // return // TRUE/FALSE //============================================== BOOL CSendRecvThread::CanSendNow(DWORD dwNow) { BOOL fRet = FALSE; if (GetdwInterval(dwNow, m_dwPrevSentTime) >= m_dwSendInterval) fRet = TRUE; return(fRet); } //============================================== // function // ★このスレッドが作ったスレッドが無効(致命的エラー)の時破棄 // 破棄したときは切断処理のために CSendRecvThreadも終わる // parameter // なし // return // TRUE:致命的エラーで破棄した/FALSE:致命的エラーなし //============================================== BOOL CSendRecvThread::KillZombei() { BOOL fRet = FALSE; if (m_pCRecvFileThread != NULL) { if (m_pCRecvFileThread->IsZombie() == TRUE) { m_pCRecvFileThread->End(); m_pCRecvFileThread->WaitForEnd(); SAFE_DELETE(m_pCRecvFileThread) fRet = TRUE; } } return(fRet); }

これで、SimpleServerTransFile(Linux)にファイル受信機能が入りました。
前回作成したSimpleClientTransFileと一緒に動かして、ファイル送受信を
やってみてください。
速度を上げて(1Mbps以上)送受信したいときは、SimpleClientTransFileの
SEND_BPSとSENDBUFSIZEを変更してください。
例
#define SEND_BPS (1000000.0 * 2.0) // 送信速度2Mbps
#define SENDBUFSIZE (1024 * 4)
このとき、パケットキャプチャソフト(WireSharkなど)で見てみると
TCPのパケットがMTUの範囲にエンドポイントが分割してくれることがわかります。

ファイル受信機能が入ったプロジェクト
SimpleServerTransFile for Linux
SimpleServerTransFile for Windows