GSI Object Oriented Online Offline (Go4)  GO4-6.1.4
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros
TGo4CommandRunnable.cxx
Go to the documentation of this file.
1 // $Id: TGo4CommandRunnable.cxx 2757 2020-04-16 11:45:21Z linev $
2 //-----------------------------------------------------------------------
3 // The GSI Online Offline Object Oriented (Go4) Project
4 // Experiment Data Processing at EE department, GSI
5 //-----------------------------------------------------------------------
6 // Copyright (C) 2000- GSI Helmholtzzentrum fuer Schwerionenforschung GmbH
7 // Planckstr. 1, 64291 Darmstadt, Germany
8 // Contact: http://go4.gsi.de
9 //-----------------------------------------------------------------------
10 // This software can be used under the license agreements as stated
11 // in Go4License.txt file which is part of the distribution.
12 //-----------------------------------------------------------------------
13 
14 #include "TGo4CommandRunnable.h"
15 
16 #include <sstream>
17 #include <iostream>
18 
19 #include "TGo4Log.h"
20 #include "TGo4Thread.h"
21 #include "TGo4BufferQueue.h"
22 #include "TGo4Socket.h"
24 #include "TGo4CommandInvoker.h"
25 #include "TGo4TaskHandler.h"
27 #include "TGo4ClientTask.h"
28 #include "TGo4ComQuit.h"
29 
30 
31 
33  TGo4TaskHandler* hand, Bool_t receivermode) :
34  TGo4TaskHandlerRunnable(name,man,hand,receivermode)
35 {
39 }
40 
41 
43 {
44 }
45 
46 Int_t TGo4CommandRunnable::Run(void* ptr)
47 {
48  if(!CheckTransportOpen()) return 0;
49  if(fbReceiverMode)
50  {
51  // wait for something from socket
52  Int_t rev=fxTransport->ReceiveBuffer();
53  if(rev>=0)
54  {
55  TBuffer* buffer=const_cast<TBuffer*> (fxTransport->GetBuffer());
56  Int_t val=0;
57  if(CheckStopBuffer(buffer,&val)) return 0; // stop for disconnect mode
58  if(val>=0)
59  {
60  fxTransport->Send(TGo4TaskHandler::Get_fgcOK()); //acknowledge before execute
61  TGo4Task* cli = dynamic_cast<TGo4Task*>(fxManager);
62  // we have a direct command, execute this here:
63  TGo4ComQuit* qcommand = 0;
65  // test here for different command values
66  switch(comvalue)
67  {
68  case kComQuit:
70  fxManager->SetBeingQuit(kTRUE); // flag for the application that we expect to be quit soon
71  TGo4Log::Debug(" Command runnable executing direct command QUIT... ");
72  //cli->Quit();
73  // note: need execution of quit in local command thread,
74  // because we are inside thread to be cancelled...
75  qcommand = new TGo4ComQuit;
76  cli->SubmitLocalCommand(qcommand);
77  break;
78  case kComKillMain:
80  TGo4Log::Debug(" Command runnable executing direct command KILL MAIN... ");
81  cli->KillMain();
82  break;
83  case kComRestartMain:
85  TGo4Log::Debug(" Command runnable executing direct command RESTART MAIN... ");
86  cli->RestartMain();
87  break;
88  case kComCloseInput:
89  // this case should be treated by CheckStopBuffer, so:
90  TGo4Log::Debug("NEVER COME HERE: Command runnable has direct command CLOSE INPUT");
91  // TGo4Log::Debug(" Command runnable executing direct command CLOSE INPUT... ");
92  // GetThread()->Stop();
93  break;
94  default:
95  TGo4Log::Debug(" Command runnable direct command value %d UNKNOWN! ",
96  comvalue);
97  break;
98 
99 
100  }
101  // end direct command section
102  } // if(val>0)
103  else
104  {
105  fxBufferQueue->AddBuffer(buffer, kTRUE);
107  } // end if((val>0))
108  } // if(rev>=0)
109  else
110  {
112  {
113  // TSocket error because of window resize, do not abort!
114  TGo4Log::Debug(" %s: caught SIGWINCH ", GetName());
116  }
117  else if(fxManager->IsTerminating())
118  {
119  //std::cout <<"Receive error in "<<GetName()<< " while threadmanager is terminating. Ignored!" << std::endl;
120  TGo4Log::Debug("Receive Error in %s during threadmanager termination. Ignored.",GetName());
121  GetThread()->Stop();
122 
123  }
124  else
125  {
126  if(fxTaskHandler->IsClientMode()) RedirectIO(); // only suppress output for analysis clients (correct shutdown if run in QtWindow!)
127  TGo4Log::Debug("Receive Error in %s, aborting taskhandler...",GetName());
128  throw TGo4TaskHandlerAbortException(this);
129  }
130  } // end if (rev>=0)
131  } //end if(fbReceiverMode)
132  else
133  {
134  // get next command from queue or wait for it
135  TBuffer* buf= fxBufferQueue->WaitBuffer();
136  if (buf)
137  // check if there is really an object from queue
138  {
139  Bool_t stopmode=CheckStopBuffer(buf);
140  // send over inter task transport
141  fxTransport->SendBuffer(buf);
142  // std::cout << "command runnable: sending buffer via transport" << std::endl;
144  if(stopmode) return 0; // no handshake after stop buffer
145  char* revchar=fxTransport->RecvRaw("dummy"); // wait for o.k. string
146  if(revchar==0)
147  {
148  // error, received garbage
150  //if (TGo4SocketSignalHandler::fgxLastSignal == kSigWindowChanged)
151  {
152  // TSocket error because of window resize, do not abort!
153  TGo4Log::Debug(" %s: caught SIGWINCH ",GetName());
155  //TGo4SocketSignalHandler::fgxLastSignal = (ESignals) 0;
156  }
157  else
158  {
159  //RedirectIO(); // do not redirect output for command sender, will not terminate!
160  TGo4Log::Debug(" !!!Connection Error -1 in CommandRunnable ''%s''!!!",GetName());
161  throw TGo4TaskHandlerAbortException(this);
162  }
163  }
164  else
165  {
166  // we received something, proceed
167  } // end if(revchar==0)
168 
169  if(!strcmp(revchar,TGo4TaskHandler::Get_fgcOK()) )
170  {
171  // fine, command has reached its destination, anyhow..
172  }
173  else if(!strcmp(revchar,TGo4TaskHandler::Get_fgcERROR()))
174  {
175  // client signals any kind of error with last command
176  TGo4Log::Debug(" CommandRunnable ''%s'' received command error string!!!",GetName());
177  }
178  else
179  {
180  // error, received something else
181  TGo4Log::Debug(" !!!Connection Error -3 in CommandRunnable ''%s''!!!",GetName());
182  //throw TGo4TerminateException(this);
183  }
184  }
185  else
186  {
187  if(!GetThread()->IsRunning())
188  {
189  // when gui client is disconnected by slave server itself
190  // (server shutdown from other gui), we have to react on Queue wakeup
191  // in command queue from the TGo4TaskHandler::StopTransportThreads
192  //std::cout <<"Command runnable gets null from queue and is stopped!!!!" << std::endl;
193  return 0;
194  }
195  // do nothing, return for next wait
196  }
197  } // end else if(fbReceiverMode)
198  return 0;
199 }
200 
201 
203 {
204  // note: this method is only called in case of aborting,
205  // so we do not care about our memory leak...JA
206  std::ostringstream* strout = new std::ostringstream;
207  //std::streambuf* ccc_buffer = std::cout.rdbuf(); // would need it to recover std::cout
208  std::cout.rdbuf(strout->rdbuf());
209  std::ostringstream* strerr = new std::ostringstream;
210  //std::streambuf* std::cerr_buffer = std::cerr.rdbuf(); // would need to recover std::cerr later
211  std::cerr.rdbuf(strerr->rdbuf());
212 }
TGo4Socket * GetCommandTransport() const
Go4EmergencyCommand_t
Bool_t SubmitLocalCommand(TGo4Command *com)
Definition: TGo4Task.cxx:529
TGo4ThreadManager * fxManager
Definition: TGo4Runnable.h:71
const TBuffer * GetBuffer() const
Definition: TGo4Socket.h:58
virtual Int_t Send(TObject *obj)
Definition: TGo4Socket.cxx:338
Int_t SendBuffer(TBuffer *buf)
Definition: TGo4Socket.cxx:222
virtual Int_t Run(void *ptr)
Bool_t Stop()
Definition: TGo4Thread.cxx:327
TGo4Thread * GetThread() const
Definition: TGo4Runnable.h:45
static const char * Get_fgcERROR()
static const char * Get_fgcOK()
Bool_t IsClientMode() const
virtual void RestartMain()
Definition: TGo4Task.cxx:126
static void SetLastSignal(Int_t v=0)
TGo4Queue * GetCommandQueue() const
TBuffer * WaitBuffer()
Int_t ReceiveBuffer()
Definition: TGo4Socket.cxx:275
static TGo4CommandInvoker * Instance()
TGo4CommandInvoker * fxInvoker
void AddBuffer(TBuffer *buffer, Bool_t clone=kFALSE)
Bool_t CheckStopBuffer(TBuffer *buf, Int_t *result=0)
Go4CommandMode_t GetRole()
virtual void KillMain()
Definition: TGo4Task.cxx:121
void SetBeingQuit(Bool_t on)
Bool_t IsTerminating() const
virtual char * RecvRaw(const char *name=0)
Definition: TGo4Socket.cxx:407
void FreeBuffer(TBuffer *buffer)
static void Debug(const char *text,...)
Definition: TGo4Log.cxx:274