source: XIOS/trunk/src/context_server.cpp @ 539

Last change on this file since 539 was 511, checked in by mhnguyen, 9 years ago

Seperating database of context on "client" side and "server" side

+) Add one more context in contex client in case of attached mode
+) Do some minor changements to make sure everything fine in case of attached mode
+) Replace buffer group with the new options

Test
+) On Curie
+) Connection mode: Attached and seperated
+) File mode: one and multiple
+) All tests passed

  • Property copyright set to
    Software name : XIOS (Xml I/O Server)
    http://forge.ipsl.jussieu.fr/ioserver
    Creation date : January 2009
    Licence : CeCCIL version2
    see license file in root directory : Licence_CeCILL_V2-en.txt
    or http://www.cecill.info/licences/Licence_CeCILL_V2-en.html
    Holder : CEA/LSCE (Laboratoire des Sciences du CLimat et de l'Environnement)
    CNRS/IPSL (Institut Pierre Simon Laplace)
    Project Manager : Yann Meurdesoif
    yann.meurdesoif@cea.fr
  • Property svn:eol-style set to native
File size: 7.1 KB
Line 
1#include "context_server.hpp"
2#include "buffer_in.hpp"
3#include "type.hpp"
4#include "context.hpp"
5#include "object_template.hpp"
6#include "group_template.hpp"
7#include "attribute_template.hpp"
8#include "domain.hpp"
9#include "field.hpp"
10#include "file.hpp"
11#include "grid.hpp"
12#include "mpi.hpp"
13#include "tracer.hpp"
14#include "timer.hpp"
15#include "cxios.hpp"
16#include "event_scheduler.hpp"
17#include "server.hpp"
18#include <boost/functional/hash.hpp>
19
20
21
22namespace xios
23{
24
25  CContextServer::CContextServer(CContext* parent,MPI_Comm intraComm_,MPI_Comm interComm_)
26  {
27    context=parent ;
28    intraComm=intraComm_ ;
29    MPI_Comm_size(intraComm,&intraCommSize) ;
30    MPI_Comm_rank(intraComm,&intraCommRank) ;
31    interComm=interComm_ ;
32    int flag ;
33    MPI_Comm_test_inter(interComm,&flag) ;
34    if (flag) MPI_Comm_remote_size(interComm,&commSize);
35    else  MPI_Comm_size(interComm,&commSize) ;
36    currentTimeLine=0 ;
37    scheduled=false ;
38    finished=false ;
39
40    boost::hash<string> hashString ;
41    hashId=hashString(context->getId()) ;
42
43  }
44  void CContextServer::setPendingEvent(void)
45  {
46    pendingEvent=true ;
47  }
48
49  bool CContextServer::hasPendingEvent(void)
50  {
51    return pendingEvent ;
52  }
53
54  bool CContextServer::eventLoop(void)
55  {
56    listen() ;
57    checkPendingRequest() ;
58    processEvents() ;
59    return finished ;
60  }
61
62  void CContextServer::listen(void)
63  {
64    int rank;
65    int flag ;
66    int count ;
67    char * addr ;
68    MPI_Status status;
69    map<int,CServerBuffer*>::iterator it;
70
71    for(rank=0;rank<commSize;rank++)
72    {
73      if (pendingRequest.find(rank)==pendingRequest.end())
74      {
75        traceOff() ;
76        MPI_Iprobe(rank,20,interComm,&flag,&status);
77        traceOn() ;
78        if (flag==true)
79        {
80          it=buffers.find(rank) ;
81          if (it==buffers.end())
82          {
83            StdSize buffSize = 0;
84            MPI_Recv(&buffSize, 1, MPI_LONG, rank, 20, interComm, &status);
85            mapBufferSize_.insert(std::make_pair(rank, buffSize));
86            it=(buffers.insert(pair<int,CServerBuffer*>(rank,new CServerBuffer(buffSize)))).first ;
87          }
88          else
89          {
90            MPI_Get_count(&status,MPI_CHAR,&count) ;
91            if (it->second->isBufferFree(count))
92            {
93              addr=(char*)it->second->getBuffer(count) ;
94              MPI_Irecv(addr,count,MPI_CHAR,rank,20,interComm,&pendingRequest[rank]) ;
95              bufferRequest[rank]=addr ;
96            }
97          }
98        }
99      }
100    }
101  }
102
103  void CContextServer::checkPendingRequest(void)
104  {
105    map<int,MPI_Request>::iterator it;
106    list<int> recvRequest ;
107    list<int>::iterator itRecv;
108    int rank ;
109    int flag ;
110    int count ;
111    MPI_Status status ;
112
113    for(it=pendingRequest.begin();it!=pendingRequest.end();it++)
114    {
115      rank=it->first ;
116      traceOff() ;
117      MPI_Test(& it->second, &flag, &status) ;
118      traceOn() ;
119      if (flag==true)
120      {
121        recvRequest.push_back(rank) ;
122        MPI_Get_count(&status,MPI_CHAR,&count) ;
123        processRequest(rank,bufferRequest[rank],count) ;
124      }
125    }
126
127    for(itRecv=recvRequest.begin();itRecv!=recvRequest.end();itRecv++)
128    {
129      pendingRequest.erase(*itRecv) ;
130      bufferRequest.erase(*itRecv) ;
131    }
132  }
133
134  void CContextServer::processRequest(int rank, char* buff,int count)
135  {
136
137    CBufferIn buffer(buff,count) ;
138    char* startBuffer,endBuffer ;
139    int size, offset ;
140    size_t timeLine ;
141    map<size_t,CEventServer*>::iterator it ;
142
143    while(count>0)
144    {
145      char* startBuffer=(char*)buffer.ptr() ;
146      CBufferIn newBuffer(startBuffer,buffer.remain()) ;
147      newBuffer>>size>>timeLine ;
148
149      it=events.find(timeLine) ;
150      if (it==events.end()) it=events.insert(pair<int,CEventServer*>(timeLine,new CEventServer)).first ;
151      it->second->push(rank,buffers[rank],startBuffer,size) ;
152
153      buffer.advance(size) ;
154      count=buffer.remain() ;
155    }
156
157  }
158
159  void CContextServer::processEvents(void)
160  {
161    map<size_t,CEventServer*>::iterator it ;
162    CEventServer* event ;
163
164    it=events.find(currentTimeLine) ;
165    if (it!=events.end())
166    {
167      event=it->second ;
168
169      if (event->isFull())
170      {
171        if (!scheduled && !CXios::isServer)
172        {
173          CServer::eventScheduler->registerEvent(currentTimeLine,hashId) ;
174          scheduled=true ;
175        }
176        else if (CXios::isServer || CServer::eventScheduler->queryEvent(currentTimeLine,hashId) )
177        {
178         CTimer::get("Process events").resume() ;
179         dispatchEvent(*event) ;
180         CTimer::get("Process events").suspend() ;
181         pendingEvent=false ;
182         delete event ;
183         events.erase(it) ;
184         currentTimeLine++ ;
185         scheduled = false ;
186        }
187      }
188    }
189  }
190
191  CContextServer::~CContextServer()
192  {
193    map<int,CServerBuffer*>::iterator it ;
194    for(it=buffers.begin();it!=buffers.end();++it) delete it->second ;
195  }
196
197
198  void CContextServer::dispatchEvent(CEventServer& event)
199  {
200    string contextName ;
201    string buff ;
202    int MsgSize ;
203    int rank ;
204    list<CEventServer::SSubEvent>::iterator it ;
205    CContext::setCurrent(context->getId()) ;
206
207    if (event.classId==CContext::GetType() && event.type==CContext::EVENT_ID_CONTEXT_FINALIZE)
208    {
209      info(20)<<"Server Side context <"<<context->getId()<<"> finalized"<<endl ;
210      std::map<int, StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
211                                             iteMap = mapBufferSize_.end(), itMap;
212      StdSize totalBuf = 0;
213      for (itMap = itbMap; itMap != iteMap; ++itMap)
214      {
215        report(10)<< " Memory report : Context <"<<context->getId()<<"> : server side : memory used for buffer of each connection to client" << endl
216                  << "  +) With client of rank " << itMap->first << " : " << itMap->second << " bytes " << endl;
217        totalBuf += itMap->second;
218      }
219      context->finalize() ;
220      finished=true ;
221      report(0)<< " Memory report : Context <"<<context->getId()<<"> : server side : total memory used for buffer "<<totalBuf<<" bytes"<<endl ;
222    }
223    else if (event.classId==CContext::GetType()) CContext::dispatchEvent(event) ;
224    else if (event.classId==CContextGroup::GetType()) CContextGroup::dispatchEvent(event) ;
225    else if (event.classId==CDomain::GetType()) CDomain::dispatchEvent(event) ;
226    else if (event.classId==CDomainGroup::GetType()) CDomainGroup::dispatchEvent(event) ;
227    else if (event.classId==CAxis::GetType()) CAxis::dispatchEvent(event) ;
228    else if (event.classId==CAxisGroup::GetType()) CAxisGroup::dispatchEvent(event) ;
229    else if (event.classId==CGrid::GetType()) CGrid::dispatchEvent(event) ;
230    else if (event.classId==CGridGroup::GetType()) CGridGroup::dispatchEvent(event) ;
231    else if (event.classId==CField::GetType()) CField::dispatchEvent(event) ;
232    else if (event.classId==CFieldGroup::GetType()) CFieldGroup::dispatchEvent(event) ;
233    else if (event.classId==CFile::GetType()) CFile::dispatchEvent(event) ;
234    else if (event.classId==CFileGroup::GetType()) CFileGroup::dispatchEvent(event) ;
235    else if (event.classId==CVariable::GetType()) CVariable::dispatchEvent(event) ;
236    else
237    {
238      ERROR("void CContextServer::dispatchEvent(CEventServer& event)",<<" Bad event class Id"<<endl) ;
239    }
240  }
241}
Note: See TracBrowser for help on using the repository browser.