PDG_Service.h

Go to the documentation of this file.

/*
 * PROPRIETARY INFORMATION.  This software is proprietary to
 * Side Effects Software Inc., and is not to be reproduced,
 * transmitted, or disclosed in any way without written permission.
 *
 * COMMENTS:
 */

#ifndef __PDG_SERVICE_H__
#define __PDG_SERVICE_H__

#include "PDG_API.h"
#include "PDG_RegisteredType.h"
#include "PDG_ServiceTypes.h"
#include "PDG_WorkItemTypes.h"

#include <PDGT/PDGT_ValueArgs.h>

#include <UT/UT_Array.h>
#include <UT/UT_ConcurrentSet.h>
#include <UT/UT_Lock.h>
#include <UT/UT_StringArray.h>
#include <UT/UT_StringHolder.h>
#include <UT/UT_UniquePtr.h>

class PDG_Scheduler;
class PY_PyObject;
class UT_Thread;
class UT_WorkBuffer;

namespace PDGN
{
    class PDGN_Message;
    class PDGN_PollingClientNNG;
}

/* 
 * Represents a registered service object, tracks the list of active clients
 * for the service, and keeps track of various settings like the service
 * pool size, MQ connection information, and service state.
 */
class PDG_API PDG_Service : public PDG_TypeInstance
{
public:
    /// Service data format version
    static const UT_StringHolder theDataVersion;

    /// Enumeration of return codes from acquireClient
    enum ServiceAcquireResult
    {
        /// A service client was successfully acquired
        eServiceAcquireSuccess,

        /// A service client was successfully acquired and locked
        eServiceAcquireSuccessLocked,

        /// All service workers are already busy
        eServiceAcquireBusy,

        /// The service worker is aleady in use by a different task
        /// with the same service lock id
        eServiceAcquireLocked,

        /// The service lock is invalid -- should never occur in practice
        eServiceAcquireInvalidLock,
    };

    /// Service client array entry
    struct ClientInfo
    {
        /// Constructs a new client info instance
                        ClientInfo(
                            const PDG_Service* service,
                            const UT_StringHolder& server_address,
                            int server_port,
                            int client_num);

        /// Constructs an incomplete client instance -- this is used for
        /// compatbility with older version of Houdini that did not
        /// explicitly track client info
                        ClientInfo(const UT_StringHolder& client_name);

        /// Returns the command for the client as a string, but combining the
        /// command arguments 
        UT_StringHolder commandString() const;

        /// Spawns the client as a child process, and returns true on
        /// success
        bool            spawnProcess();

        /// /// Kills the active client process, if one exists
        void            killProcess();


        /// The array of command line args for the client
        UT_StringArray  myCommandArgs;

        /// The name of the service client
        UT_StringHolder myName;

        /// The log path for client output
        UT_StringHolder myLogPath;

        /// The connection address
        UT_StringHolder myAddress;


        /// The connection port
        exint           myPort;

        /// The process or job id for the client
        exint           myId;

        /// The total number of tasks the service has cooked
        exint           myTotalCooks;

        /// The amount of memory in use by the process, as recorded by the
        /// last task that cooked using the client
        int64           myMemoryUsage;

        /// The ID of the work item that has locked the service client, if
        /// any
        PDG_WorkItemID  myLock;

        /// Set to true once the client has connected
        bool            myIsConnected;

        /// Set to true if the client is active for a job
        bool            myIsActive;
    };

public:

    /// Constructs a new service instance with a default pool size of 1, and
    /// no connection information.
                            PDG_Service(
                                const PDG_BaseType* type,
                                const PDGT_ValueArgs& extra_args,
                                const UT_StringHolder& service_command,
                                bool internal_service);

                            ~PDG_Service() override;

    /// Sets/gets the unique name for the service. The name of the service is
    /// unique to the PDG_ServiceManager that created it.
    void                    setName(const UT_StringHolder& name)
                                { myName = name; }
    const UT_StringHolder&  name() const
                                { return myName; }

    /// Returns true if the service is internal to PDG. Internal services
    /// cannot be directly managed by users.
    bool                    isInternal() const
                                { return myIsInternal; }

    /// Sets/gets the persistent flag, which determines whether a service
    /// should be persisted to the pdgservice JSON configuration file. A
    /// service that isn't persistent will be lost when the Houdini session
    /// is closed.
    void                    setPersistent(bool persistent)
                                { myIsPersistent = persistent; }
    bool                    isPersistent() const
                                { return myIsPersistent; }

    /// Returns true if the service has errors when starting
    bool                    hasErrors() const
                                { return myHasErrors; }

    /// Sets/gets the autostart flag, which indicates that the service will
    /// start automatically when work items need to use it.
    void                    setAutoStart(bool auto_start)
                                { myIsAutoStart = auto_start; }
    bool                    isAutoStart() const
                                { return myIsAutoStart; }

    /// Sets/gets an opaque Python data object that can be stored on the
    /// service.
    void                    setData(PY_PyObject* new_data);
    PY_PyObject*            data()
                                { return myData; }

    /// Sets/gets the MQ url that the service should use to relay messages
    /// between service clients and the PDG graph
    void                    setMqUrl(const UT_StringHolder& url)
                                { myMqUrl = url; }
    const UT_StringHolder&  mqUrl() const
                                { return myMqUrl; }

    /// Sets/gets the log level for the MQ process that relays messages
    /// between clients and the PDG graph
    void                    setMqLogLevel(int log_level)
                                { myMqLogLevel = log_level; }
    int                     mqLogLevel() const
                                { return myMqLogLevel; }

    /// Sets/gets the port used by the MQ relay.
    void                    setMqPort(int mq_port)
                                { myMqPort = mq_port; }
    int                     mqPort() const
                                { return myMqPort; }

    /// Sets/gets the directory that service clients should write their log
    /// files to.
    void                    setClientLogDir(const UT_StringHolder& log_dir)
                                { myClientLogDir = log_dir; }
    const UT_StringHolder&  clientLogDir() const
                                { return myClientLogDir; }

    /// Sets/gets the directory that the MQ relay writes its logs to.
    void                    setMqLogDir(const UT_StringHolder& log_dir)
                                { myMqLogDir = log_dir; }
    const UT_StringHolder&  mqLogDir() const
                                { return myMqLogDir; }

    /// Sets how the service itself should log output
    void                    setClientLogType(PDG_ServiceLogType log_type)
                                { myClientLogType = log_type; }
    PDG_ServiceLogType      clientLogType() const
                                { return myClientLogType; }

    /// Sets/gets the names for the environment variables to be set in 
    /// this job's environment. This only applies if the service does not
    /// have a scheduler.
    void                    setEnvVarNames(const UT_StringArray& env_var_names)
                                { myEnvVarNames = env_var_names; }
    const UT_StringArray&   envVarNames() const
                                { return myEnvVarNames; }

    /// Sets/gets the values for the environment variables to be set in 
    /// this job's environment. This only applies if the service does not
    /// have a scheduler.
    void                    setEnvVarValues(const UT_StringArray& env_var_values)
                                { myEnvVarValues = env_var_values; }
    const UT_StringArray&   envVarValues() const
                                { return myEnvVarValues; }

    /// Sets/gets the extra args array for the service
    void                    setExtraArguments(const UT_StringArray& args)
                                { myExtraArguments = args; }
    const UT_StringArray&   extraArguments() const
                                { return myExtraArguments; }

    /// For scheduler services, sets the name of the scheduler and graph
    /// that created the service.
    void                    setScheduler(
                                const UT_StringHolder& name,
                                const UT_StringHolder& context_name);

    /// Clears the scheduler information associated with a scheduler
    /// service
    void                    clearScheduler();

    /// Returns the name of the scheduler associated with the service, if
    /// it's a scheduler service
    const UT_StringHolder&  schedulerName() const
                                { return mySchedulerName; }

    /// Returns the name of the graph that owns the scheduler that created
    /// the service, if it's a scheduler service
    const UT_StringHolder&  schedulerContextName() const
                                { return mySchedulerContextName; }

    /// Sets/gets the service owner type -- either a session service or a
    /// scheduler service
    void                    setOwner(PDG_ServiceOwner owner)
                                { myServiceOwner = owner; }
    PDG_ServiceOwner        owner() const
                                { return myServiceOwner; }

    /// Sets/gets the current state of the service
    void                    setState(PDG_ServiceState state)
                                { myState = state; }
    PDG_ServiceState        state() const
                                { return myState; }

    /// Sets/gets the command line string that should be used to start clients
    /// running in the service pool
    void                    setCommand(const UT_StringHolder& command)
                                { myCommand = command; }
    const UT_StringHolder&  command() const
                                { return myCommand; }

    /// Sets/gets the pool size for the service, which determines the number
    /// of concurrent client processes that service executes
    void                    setPoolSize(int pool_size)
                                { myPoolSize = pool_size; }
    int                     poolSize() const
                                { return myPoolSize; }

    /// Sets/gets the connection port for the service itself
    void                    setPort(int port)
                                { myPort = port; }
    int                     port() const
                                { return myPort; }

    /// Sets the service connection timeout in milliseconds
    void                    setConnectionTimeout(int ms)
                                { myConnectionTimeout = ms; }
    int                     connectionTimeout() const
                                { return myConnectionTimeout; }

    /// Sets the memory limit and reset type for the service
    void                    setMemoryResetType(
                                    PDG_ServiceResetType memory_reset_type)
                                {
                                    myMemoryResetType = memory_reset_type;
                                }
    void                    setMemoryLimit(
                                    exint memory_limit)
                                {
                                    myMemoryLimit = memory_limit;
                                }
    PDG_ServiceResetType    memoryResetType() const
                                { return myMemoryResetType; }
    exint                   memoryLimit() const
                                { return myMemoryLimit; }

    /// Returns true if the service is start or is already running
    bool                    isStartingOrRunning() const;

    /// Returns true if the service is stopping, or has already stopped
    bool                    isStoppingOrStopped() const;

    /// Queries and returns the scheduler associated with the service, if the
    /// service is a scheduler service
    PDG_Scheduler*          serviceScheduler(UT_WorkBuffer& errors) const;


    /// Queries the list of all active service client names
    UT_StringArray          clientNames() const;

    /// Queries the name of a particular service client, using its client
    /// number
    UT_StringHolder         clientName(int client_num) const;

    /// Returns the service client info for a particular client
    ClientInfo*             client(const UT_StringHolder& name);

    /// Adds a pending service client, which can be waiting on using
    /// waitForClients
    ClientInfo*             addPendingClient(
                                const UT_StringHolder& server_address,
                                int server_port,
                                int client_num);


    /// Stats a service pool. If background is true the service processes are
    /// start asynchronously, and control immediately returns after calling
    /// this function. It's up to the caller to ensure that the service has
    /// started before using it. Returns either success, failure, or already-
    /// running.
    PDG_ServiceStartResult  startService(
                                UT_WorkBuffer& errors,
                                bool background);

    /// Resets a specific service client, or all clients if empty string is
    /// passed in as the client_name
    bool                    resetClient(
                                UT_WorkBuffer& errors,
                                const UT_StringHolder& client_name =
                                    UT_StringHolder::theEmptyString);

    /// Restarts a service client. If client_name is empty string all
    /// clients are restarted.
    bool                    restartClient(
                                UT_WorkBuffer& errors,
                                const UT_StringHolder& client_name);

    /// Stops a service client and kills the process associated with it.
    /// The client can be started again at a later point using restartClient.
    /// If client_name is empty string all clients are stoppped.
    bool                    stopClient(
                                UT_WorkBuffer& errors,
                                const UT_StringHolder& client_name);

    /// Immediately kills the process associated with a service client.
    /// The client can be started again at a later point using restartClient.
    /// If client_name is empty string all clients are stoppped.
    bool                    killClient(
                                UT_WorkBuffer& errors,
                                const UT_StringHolder& client_name);

    /// Stops the service if it's running, and returns a boolean to indiate 
    /// the status.
    bool                    stopService(
                                UT_WorkBuffer& errors,
                                bool ignore_stopped);

    /// Kills the service if it's running, and returns a boolean to indiate 
    /// the status.
    bool                    killService(
                                UT_WorkBuffer& errors,
                                bool ignore_stopped);

    /// Cleans up the clients in the service. This method is public so that
    /// schedulers can call it on services they manage.
    bool                    cleanupService(
                                UT_WorkBuffer& errors,
                                bool stop_mq_server,
                                bool wait_for_clients);
    
    /// Submits a work item to a service client asynchronously, and returns
    /// the pending message handle for the job.
    PDGN::PDGN_Message*     executeWorkItem(
                                UT_WorkBuffer& errors,
                                const UT_StringHolder& client_name,
                                const PDG_WorkItem* work_item);

    /// Submits a work item to a service client asychronously, and returns
    /// the pending messae handle for the job. This variant of the method
    /// allows the caller to supply a custom script that will run instead of
    /// the standard job script or script data associated with the work item.
    PDGN::PDGN_Message*     executeWorkItem(
                                UT_WorkBuffer& errors,
                                const UT_StringHolder& client_name,
                                const PDG_WorkItem* work_item,
                                const UT_WorkBuffer& script_buffer);

    /// Starts a background thread that polls the ready listener until all
    /// service clients have connected
    void                    startPollingClient();

    /// This will force stop the ready listener. If you want to wait for it to
    /// complete, you should call waitForClients() instead.
    void                    stopPollingClient();

    /// Returns the name of the polling client, if one exists
    const UT_StringHolder&  pollingClientName() const;

    /// Returns true if the service has an active polling client
    bool                    hasPollingClient() const
                                { return (myPollingClient != nullptr); }

    // Returns a pointer to the polling client, or nullptr if none exists.
    PDGN::PDGN_PollingClientNNG* 
                            getPollingClient() const
                                { return myPollingClient.get(); }

    /// Waits for all pending service clients to connect, and then terminates
    /// the ready listener
    void                    waitForClients();

    /// Increments the number of actively starting client
    void                    incrementStartingClientsCount();

    /// Returns the command for the service client with the specified client
    /// number, as a string.
    UT_StringHolder         getCommand(
                                const char* server_address,
                                int server_port,
                                int client_num) const;

    /// Returns the command for the service client with the specified client
    /// number, as an array of arguments
    void                    getCommand(
                                UT_StringArray& command_tokens,
                                const char* server_address,
                                int server_port,
                                int client_num) const;

    /// Pings the specified client, and returns true if the client was
    /// reachable
    bool                    ping(int client_num, UT_WorkBuffer& errors) const;

    /// Acquires a client, which reserves it for the caller
    ServiceAcquireResult    acquireClient(
                                UT_StringHolder& client_name,
                                PDG_WorkItemID item_lock);

    /// Releases a client that was previously acquired, but does not unlock it
    bool                    releaseClient(
                                UT_WorkBuffer& errors,
                                const UT_StringHolder& client_name,
                                PDG_ServiceResetWhen reset_when,
                                PDG_ServiceResetType reset_type,
                                int64 memory_usage);

    /// Unlocks a service client
    bool                    unlockClient(
                                PDG_WorkItemID item_lock,
                                PDG_ServiceResetType reset_type,
                                UT_WorkBuffer& errors);

private:
    enum ClientCleanupType
    {
        eClientCleanupNone,
        eClientCleanupWait,
        eClientCleanupKill
    };

    static void*            runSpawnSessionService(void* param);
    static void*            runPollingThread(void* param);

    bool                    getClientNameFromMessage(
                                PDGN::PDGN_Message* msg,
                                UT_StringHolder& client_name);

    bool                    startSessionService(
                                UT_WorkBuffer& errors,
                                bool background);
    bool                    spawnSessionServiceClients();
    void                    pollClients();

    bool                    stopService(
                                UT_WorkBuffer& errors,
                                ClientCleanupType cleanup_type,
                                bool ignore_stopped);
    bool                    stopSessionService(
                                UT_WorkBuffer& errors,
                                ClientCleanupType cleanup_type);
    bool                    stopClient(
                                UT_WorkBuffer& errors,
                                int client_index,
                                bool restart);
    bool                    killClient(
                                UT_WorkBuffer& errors,
                                int client_index,
                                bool restart);
    bool                    resetClient(
                                UT_WorkBuffer& errors,
                                ClientInfo* client);
    bool                    cleanupClients(
                                UT_WorkBuffer& errors,
                                ClientCleanupType cleanup_type,
                                bool stop_mq_server);

private:
    static constexpr int    theMaxConcurrentStartup = 24;
    static constexpr int    theSleepTime = 10;
    static constexpr int    theMQSleepTime = 100;

    using PollingClient     = UT_UniquePtr<PDGN::PDGN_PollingClientNNG>;
    using ClientInfoArray   = UT_Array<UT_UniquePtr<ClientInfo>>;

private:
    ClientInfoArray         myPendingClients;
    ClientInfoArray         myActiveClients;
    PDG_WorkItemIDSet       myActiveLocks;
    
    UT_StringHolder         myName;
    UT_StringHolder         myCommand;

    UT_StringHolder         mySchedulerName;
    UT_StringHolder         mySchedulerContextName;

    UT_StringHolder         myMqUrl;
    UT_StringHolder         myMqLogDir;

    UT_StringHolder         myClientLogDir;

    UT_StringArray          myEnvVarNames;
    UT_StringArray          myEnvVarValues;
    UT_StringArray          myExtraArguments;

    PollingClient           myPollingClient;
    UT_Thread*              myPollingThread;
    UT_Thread*              myStartThread;

    PY_PyObject*            myData;

    exint                   myMemoryLimit;

    pid_t                   myMqPid;

    SYS_AtomicInt<int>      myPollingClientCount;

    int                     myConnectionTimeout;
    int                     myMqLogLevel;
    int                     myMqPort;
    int                     myPoolSize;
    int                     myPort;

    PDG_ServiceOwner        myServiceOwner; 
    PDG_ServiceState        myState;
    PDG_ServiceLogType      myClientLogType;

    PDG_ServiceResetType    myMemoryResetType;

    bool                    myIsAutoStart;
    bool                    myIsInternal;
    bool                    myIsPersistent;
    bool                    myHasErrors;

    UT_Lock                 myServiceLock;
    mutable UT_Lock         myClientLock;
};

#endif