proactor传文件的测试

wishel

最初楼主的程序我看了几眼就看不下去了，问题很多，有的还很低级，比如读到的文件内容还在内存里进行了一次多余的copy。modern是个好同志，非常有耐心帮你看完并指出问题提出建议。
我劝楼主不要急躁，别贪多，循序渐进把基础打扎实。ace的学习曲线很陡，基础不好还要贪多的话肯定容易晕。多花点时间打基础，绝不会拖慢你学ace，开发技术除了ace还有很多，这些基础对你学其他的东西也是同样需要的。
以上言论如有冒犯请楼主不要生气哈，也别气馁，学ace的都是这么硬啃苦熬过来的。多看书，深刻理解，别人可以帮你分析解决问题但不可能代替你进行理解和记忆。如果ace书上的内容很难看懂就先找更基础一点的书看。

psycheqiqi

谢谢modern，wishel的指点
刚被老大找了谈话。说我做的太慢了。唉！！！

哪有讲做一个message block的pool的例子，百度了半天也没发现。
是不是可以用ace_message_queue来处理？

[ 本帖最后由 psycheqiqi 于 2009-10-23 11:30 编辑 ]

psycheqiqi

我不知道怎么做message_block的pool。关于message_bolck的pool哪些地方有讲？

writeDate()函数中是1450号错误。


能用meaage_queue来处理吗？
我想先初始化
private：
     ACE_Message_Queue<ACE_NULL_SYNCH> *mq_;
然后在WriteDate()中判断
if(mq_->enqueue_tail(smb)!=-1)能否加入队列中，若能加入则发送，否则循环等待。

这样能实现流量控制吗？

[ 本帖最后由 psycheqiqi 于 2009-10-23 15:31 编辑 ]

wishel

呵呵，你可以先百度一下内存池，有很多例子
message block的pool原理是类似的

急着用的话，可以用一个整数比如count=100作为上限，初始为0.每次读文件并new message block的话，count++，（读之前先判断如果count=100就什么也不做直接返回）。每次handle_write_stream()中release后，count--。

wishel

原帖由 psycheqiqi 于 2009-10-23 14:42 发表
我不知道怎么做message_block的pool。关于message_bolck的pool哪些地方有讲？

writeDate()函数中是1450号错误。


能用meaage_queue来处理吗？
我想先初始化
private：
     ACE_Message_Queue *mq_;
然后在WriteDate()中 ...

千万别循环等待，因为如果单cpu只有一个线程，循环了就完了，proactor的event loop没机会跑了。
在handle_write_stream()中调用read，read中判断是否满，满了就什么也不做返回，不满的话就一直循环读到满，每一次读完（handle_read_file）都要调用一次write。

[ 本帖最后由 wishel 于 2009-10-23 16:23 编辑 ]

psycheqiqi

假设当>100的时候我return了，return后返回失败，但返回还是会继续调用WriteDate（）。
int WriteDate(char *szSendBuf, int nSendCnt)
        {
                ACE_Message_Block *smb = new ACE_Message_Block(nSendCnt+1);
                count++;
                if(count>100)
                        return 0;
                smb->copy(szSendBuf,nSendCnt+1);                     

                int  nResult= this->writer_.write(*smb,nSendCnt+1);               
               
                if ( nResult != 0)
                {               
                        CPserverDlg *dlg=(CPserverDlg*)AfxGetApp()->GetMainWnd();
                        ACE_OS::sleep(0);
               ACE_TRACE("Write data failed!");        
                }
               
                return nResult;
        }

结果只发送那100段。是不是需要用信号量来阻塞下。

[ 本帖最后由 psycheqiqi 于 2009-10-23 16:52 编辑 ]

wishel

原帖由 psycheqiqi 于 2009-10-23 16:43 发表
假设当>100的时候我return了，return后返回失败，但返回还是会继续调用WriteDate（）。
int WriteDate(char *szSendBuf, int nSendCnt)
        {
                ACE_Message_Block *smb = new ACE_Message_Block(nSendCnt ...

WriteDate()是在handle_write_stream()中调用的，此时count已经--了，所以可以继续。

handle_write_stream(){
....
mb.relase();
count--;
init_read();// 发起一次异步读操作，调用到WriteDate()
...
}

[ 本帖最后由 wishel 于 2009-10-23 17:38 编辑 ]

psycheqiqi

我的WriteDate()不是在handle_write_stream()中调用的，而是单独定义的函数。异步的读是在对话框的OnFileSend（）函数中调用的writeDate（）
        virtual void handle_write_stream
                (const ACE_Asynch_Write_Stream::Result &result)
        {
                ACE_Message_Block &mb = result.message_block ();
                mb.release();
                count--;
                return;
        }

        void WriteDate(char *szSendBuf, int nSendCnt)
        {
                ACE_Message_Block *smb = new ACE_Message_Block(nSendCnt+1);
            count++;
                smb->copy(szSendBuf,nSendCnt+1);                     

                int  nResult= this->writer_.write(*smb,nSendCnt+1);               
               
                if ( nResult != 0)
                {               
                        CPserverDlg *dlg=(CPserverDlg*)AfxGetApp()->GetMainWnd();
                        ACE_OS::sleep(0);
               ACE_TRACE("Write data failed!");        
                }
               
                return ;
        }


void CPserverDlg::OnFileSend()
{
// TODO: Add your control notification handler code here
ACE_Time_Value delay;
delay.msec(1);
CString str;
sendFile.Abort();
    sendFile.Open(sendFileName,CFile::modeRead  | CFile::typeBinary);
sendFileLen = sendFile.GetLength();
CPserverDlg *dlg = (CPserverDlg*)AfxGetApp()->GetMainWnd();
str.Format("%d",sendFileLen);
dlg->MessageBox(str);

int size=8100;
while(sendFileLen>size)
{
  char *membuf=new char[size+2] ;
  sendFile.Read(membuf,size);
  membuf[size] = '%';
  membuf[size+1] = '\0';
  list <HA_Proactive_Service *> ::iterator its;
  for(its=HA_Proactive_Service::List.begin();its!=HA_Proactive_Service::List.end();its++)
  {

   HA_Proactive_Service *curSvr = *its;
   curSvr->WriteDate( membuf,size+1);
  }
  sendFileLen = sendFileLen-size;
  delete []membuf;
  membuf = NULL;
}
    CHAR *membuf = new CHAR[sendFileLen+2];
sendFile.Read(membuf,sendFileLen);
sendFile.Close();
membuf[sendFileLen] = '^';
membuf[sendFileLen+1] = '\0';
    list <HA_Proactive_Service *> ::iterator its;
for(its=HA_Proactive_Service::List.begin();its!=HA_Proactive_Service::List.end();its++)
{

  HA_Proactive_Service *curSvr = *its;
  curSvr->WriteDate(membuf,size+1);
}

dlg->MessageBox((LPCTSTR)"文件已发送");

delete []membuf;
     membuf = NULL;
  return 0;
}

[ 本帖最后由 psycheqiqi 于 2009-10-23 17:47 编辑 ]

psycheqiqi

谢谢wishel，办公室蚊子太多，顶不住了。先回去了。

wishel

原帖由 psycheqiqi 于 2009-10-23 17:43 发表
我的WriteDate()不是在handle_write_stream()中调用的，而是单独定义的函数。异步的读是在对话框的OnFileSend（）函数中调用的writeDate（）
        virtual void handle_write_stream
                (const ACE_Asynch_Wri ...

把OnFileSend() 中的while循环内容提炼出async_read_file();
static ACE_Atomic_Op<ACE_Thread_Mutex, unsigned long> count;
OnFileSend() { //参照test_proactor.cpp中Sender::open (const ACE_TCHAR *host, u_short port)
...
  while (count <100) {
    async_read_file();
  }
...
}
async_read_file() { //参照test_proactor.cpp中Sender::initiate_read_file (void)
...
  ACE_ASSERT(count<100);
  new mb，异步读入一片文件内容;
  ++count;
...
}
handle_read_file() { //参照test_proactor.cpp
...
  移动文件指针;
  WriteDate(mb); //发起异步写
...
}
handle_write_stream(){ //参照test_proactor.cpp
....
  mb.relase();
  --count;
  async_read_file();
...
}