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_Lock.h>
#include <UT/UT_Options.h>
#include <UT/UT_StringArray.h>
#include <UT/UT_StringHolder.h>
#include <UT/UT_UniquePtr.h>

class PDG_GraphContext;
class PDG_Scheduler;
class PDG_ServiceClient;
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:
    /// 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,
                                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; }

    /// 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 the custom data dictionary on the service.
    void                    setData(const UT_OptionsHolder& new_data)
                                { myData = new_data; }
    const UT_OptionsHolder& 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; }

    /// Returns the command line string for the service
    const UT_StringHolder&  command() const;

    /// Returns the script file path for the service
    const UT_StringHolder&  script()  const;

    /// Sets the service scheduler type
    void                    setServiceSchedulerType(
                                PDG_ServiceSchedulerType type);

    /// Sets the scheduler type name
    void                    setSchedulerTypeName(
                                const UT_StringHolder& type_name);

    /// 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 type of scheduler service in use
    PDG_ServiceSchedulerType
                            serviceSchedulerType() const
                                { return myServiceSchedulerType; }

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

    /// 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 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;

    /// Returns true if the service is using the specified scheduler
    bool                    isRunningOnScheduler(
                                const PDG_Scheduler* scheduler);

    /// Returns the scheduler that the service is running on, if any
    PDG_Scheduler*          runningScheduler() 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
    PDG_ServiceClient*      client(const UT_StringHolder& name);

    /// Adds a pending service client, which can be waiting on using
    /// waitForClients
    PDG_ServiceClient*      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,
                                const PDG_GraphContext* context,
                                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);

    /// Resumes any service clients that were stopped previously and returns
    /// a boolean to indicate the status. Has no effect on clients that are
    /// already running.
    bool                    resumeService(
                                UT_WorkBuffer& errors,
                                const PDG_GraphContext* context);


    /// 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,
                                bool as_native,
                                bool apply_wedges);

    /// 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,
                                bool as_native,
                                bool apply_wedges);

    /// Starts a background thread that polls the ready listener until all
    /// service clients have connected. If a graph context is passed in, the
    /// polling client will cancel automatically if the cook associated with
    /// the context is canceled
    void                    startPollingClient(
                                const PDG_GraphContext* context);

    /// Stops the ready listener. If wait is true then the method will block
    /// until pending clients have started, otherwise it will forcible stop
    /// the polling client
    void                    stopPollingClient(bool wait);

    /// 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(); }

    /// 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;

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

private:
    friend class PDG_ServiceScheduler;
    friend class PDG_ServiceManager;

    enum ClientCleanupType
    {
        eClientCleanupNone,
        eClientCleanupWait,
        eClientCleanupKill
    };

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

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

    bool                    startSessionService(
                                UT_WorkBuffer& errors,
                                const PDG_GraphContext* context,
                                bool background);
    bool                    spawnSessionServiceClients(
                                UT_WorkBuffer& errors,
                                const PDG_GraphContext* context);

    void                    pollClients();
    void                    waitClients();

    bool                    stopService(
                                UT_WorkBuffer& errors,
                                ClientCleanupType cleanup_type,
                                bool ignore_stopped);
    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,
                                PDG_ServiceClient* client);
    bool                    cleanupClients(
                                UT_WorkBuffer& errors,
                                ClientCleanupType cleanup_type,
                                bool stop_mq_server);

    PDG_ServiceAcquireResult
                            acquireClient(
                                UT_StringHolder& client_name,
                                PDG_WorkItemID item_lock);
    bool                    releaseClient(
                                UT_WorkBuffer& errors,
                                const UT_StringHolder& client_name,
                                PDG_ServiceResetWhen reset_when,
                                PDG_ServiceResetType reset_type,
                                int64 memory_usage);
    bool                    unlockClient(
                                PDG_WorkItemID item_lock,
                                PDG_ServiceResetType reset_type,
                                UT_WorkBuffer& errors);
    bool                    killClientProcess(
                                UT_WorkBuffer& errors,
                                PDG_ServiceClient* client);
    bool                    restartClientProcess(
                                UT_WorkBuffer& errors,
                                int client_index);

    PDG_Scheduler*          configureServiceScheduler(
                                const PDG_GraphContext* context,
                                UT_WorkBuffer& errors);
    PDG_Scheduler*          serviceScheduler(
                                UT_WorkBuffer& errors) const;

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

    using PollingClientPtr  = UT_UniquePtr<PDGN::PDGN_PollingClientNNG>;

private:
    PDG_ServiceClientArray  myPendingClients;
    PDG_ServiceClientArray  myActiveClients;
    PDG_ServiceClientArray  myCleanupClients;
    PDG_WorkItemIDSet       myActiveLocks;
    
    UT_StringHolder         myName;
    UT_StringHolder         mySchedulerTypeName;
    UT_StringHolder         mySchedulerName;
    UT_StringHolder         mySchedulerContextName;

    UT_StringHolder         myRunningSchedulerName;
    UT_StringHolder         myRunningSchedulerContextName;

    UT_StringHolder         myMqUrl;
    UT_StringHolder         myMqLogDir;
    UT_StringHolder         myClientLogDir;

    UT_StringArray          myEnvVarNames;
    UT_StringArray          myEnvVarValues;
    UT_StringArray          myExtraArguments;

    PollingClientPtr        myPollingClient;
    const PDG_GraphContext* myPollingContext;
    UT_Thread*              myPollingThread;
    UT_Thread*              myStartThread;
    UT_Thread*              myCleanupThread;

    UT_OptionsHolder        myData;

    exint                   myMemoryLimit;

    pid_t                   myMqPid;

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

    PDG_ServiceOwner        myServiceOwner; 
    PDG_ServiceSchedulerType
                            myServiceSchedulerType;
    PDG_ServiceState        myState;
    PDG_ServiceLogType      myClientLogType;

    PDG_ServiceResetType    myMemoryResetType;

    bool                    myIsAutoStart;
    bool                    myIsInternal;
    bool                    myIsPersistent;

    UT_Lock                 myServiceLock;
    mutable UT_Lock         myClientLock;
};

#endif