include/root/TMPWorker.h

0001 /* @(#)root/multiproc:$Id$ */
0002 // Author: Enrico Guiraud July 2015
0003
0004 /*************************************************************************
0005  * Copyright (C) 1995-2000, Rene Brun and Fons Rademakers.               *
0006  * All rights reserved.                                                  *
0007  *                                                                       *
0008  * For the licensing terms see $ROOTSYS/LICENSE.                         *
0009  * For the list of contributors see $ROOTSYS/README/CREDITS.             *
0010  *************************************************************************/
0011
0012 #ifndef ROOT_TMPWorker
0013 #define ROOT_TMPWorker
0014
0015 #include "MPCode.h"
0016 #include "MPSendRecv.h" //MPCodeBufPair
0017 #include "PoolUtils.h"
0018 #include "TSysEvtHandler.h" //TFileHandler
0019
0020 #include <memory> //unique_ptr
0021 #include <string>
0022 #include <sstream>
0023 #include <unistd.h> //pid_t
0024
0025 class TMPWorker {
0026 public:
0027    TMPWorker() : fNWorkers(0), fMaxNEntries(0),
0028                  fProcessedEntries(0), fS(), fPid(0), fNWorker(0) { }
0029    TMPWorker(unsigned nWorkers, ULong64_t maxEntries)
0030                : fNWorkers(nWorkers), fMaxNEntries(maxEntries),
0031                  fProcessedEntries(0), fS(), fPid(0), fNWorker(0) { }
0032    virtual ~TMPWorker() { }
0033    //it doesn't make sense to copy a TMPWorker (each one has a uniq_ptr to its socket)
0034    TMPWorker(const TMPWorker &) = delete;
0035    TMPWorker &operator=(const TMPWorker &) = delete;
0036
0037    virtual void Init(int fd, unsigned workerN);
0038    void Run();
0039    TSocket *GetSocket() { return fS.get(); }
0040    pid_t GetPid() { return fPid; }
0041    unsigned GetNWorker() const { return fNWorker; }
0042
0043 protected:
0044    std::string fId; ///< identifier string in the form `W<nwrk>|P<proc id>`
0045    unsigned fNWorkers; ///< the number of workers spawned
0046    ULong64_t fMaxNEntries; ///< the maximum number of entries to be processed by this worker
0047    ULong64_t fProcessedEntries; ///< the number of entries processed by this worker so far
0048
0049    void   SendError(const std::string& errmsg, unsigned int code = MPCode::kError);
0050
0051 private:
0052    virtual void HandleInput(MPCodeBufPair &msg);
0053
0054    std::unique_ptr<TSocket> fS; ///< This worker's socket. The unique_ptr makes sure resources are released.
0055    pid_t fPid; ///< the PID of the process in which this worker is running
0056    unsigned fNWorker; ///< the ordinal number of this worker (0 to nWorkers-1)
0057 };
0058
0059 #endif