%-*-Mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*- % ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et nomod: %%% %%%------------------------------------------------------------------------ %%% @doc %%% ==CloudI Spawn== %%% @end %%% %%% MIT License %%% %%% Copyright (c) 2011-2023 Michael Truog %%% %%% Permission is hereby granted, free of charge, to any person obtaining a %%% copy of this software and associated documentation files (the "Software"), %%% to deal in the Software without restriction, including without limitation %%% the rights to use, copy, modify, merge, publish, distribute, sublicense, %%% and/or sell copies of the Software, and to permit persons to whom the %%% Software is furnished to do so, subject to the following conditions: %%% %%% The above copyright notice and this permission notice shall be included in %%% all copies or substantial portions of the Software. %%% %%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR %%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, %%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE %%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER %%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING %%% FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER %%% DEALINGS IN THE SOFTWARE. %%% %%% @author Michael Truog %%% @copyright 2011-2023 Michael Truog %%% @version 2.0.6 {@date} {@time} %%%------------------------------------------------------------------------ -module(cloudi_core_i_spawn). -author('mjtruog at protonmail dot com'). %% external interface -export([start_internal/18, start_external/21, status_internal/4, status_external/6, update_external/3, update_internal_f/9, update_external_f/12]). -include("cloudi_logger.hrl"). -include("cloudi_core_i_configuration.hrl"). -include("cloudi_core_i_constants.hrl"). -define(PROCESS_START_INTERNAL, cloudi_core_i_services_internal_sup:process_start). -define(PROCESS_START_EXTERNAL, cloudi_core_i_services_external_sup:process_start). % environmental variables used by CloudI API initialization -define(ENVIRONMENT_THREAD_COUNT, "CLOUDI_API_INIT_THREAD_COUNT"). -define(ENVIRONMENT_PROTOCOL, "CLOUDI_API_INIT_PROTOCOL"). -define(ENVIRONMENT_BUFFER_SIZE, "CLOUDI_API_INIT_BUFFER_SIZE"). % environmental variable process constants provided before initialization -define(ENVIRONMENT_TIMEOUT_INITIALIZE, "CLOUDI_API_INIT_TIMEOUT_INITIALIZE"). -define(ENVIRONMENT_TIMEOUT_TERMINATE, "CLOUDI_API_INIT_TIMEOUT_TERMINATE"). -define(ENVIRONMENT_PROCESS_INDEX, "CLOUDI_API_INIT_PROCESS_INDEX"). -define(ENVIRONMENT_PROCESS_COUNT_MAX, "CLOUDI_API_INIT_PROCESS_COUNT_MAX"). -define(ENVIRONMENT_PROCESS_COUNT_MIN, "CLOUDI_API_INIT_PROCESS_COUNT_MIN"). -ifdef(CLOUDI_CORE_STANDALONE). -dialyzer({no_match, [start_external_spawn/25, update_external/3]}). -endif. -type error_reason_start_internal() :: {service_internal_module_not_loaded, any()}. -type error_reason_start_external() :: {service_external_file_path_invalid_expanded | service_external_file_path_invalid_unicode | service_external_args_invalid_quotes | service_external_args_invalid_unicode | service_external_env_invalid_expanded | service_external_env_invalid_unicode, any()}. -export_type([error_reason_start_internal/0, error_reason_start_external/0]). % Service execution arguments that are constant across all % service instance processes (used for each start/restart). % Some argument values could be explicitly changed with an update. -type arguments_execution_internal() :: nonempty_list(pid() | % GroupLeader atom() | % Module cloudi_service_api:args_internal() | % Args cloudi_service_api: timeout_initialize_milliseconds() | % Timeout cloudi:service_name_pattern() | % Prefix cloudi_service_api: timeout_send_async_value_milliseconds() | % TimeoutAsync cloudi_service_api: timeout_send_sync_value_milliseconds() | % TimeoutSync cloudi_service_api: timeout_terminate_milliseconds() | % TimeoutTerm cloudi_service_api:dest_refresh() | % DestRefresh list(cloudi:service_name_pattern()) | undefined | % DestListDeny list(cloudi:service_name_pattern()) | undefined | % DestListAllow #config_service_options{} | % ConfigOptions cloudi_service_api:service_id()). % ID -type arguments_execution_external() :: nonempty_list(pos_integer() | % ThreadCount cloudi_service_api:file_path() | % Filename cloudi_service_api:args_external() | % Arguments cloudi_service_api:env_external() | % Environment 'local' | 'tcp' | 'udp' | % Protocol pos_integer() | % BufferSize cloudi_service_api: timeout_initialize_milliseconds() | % Timeout cloudi:service_name_pattern() | % Prefix cloudi_service_api: timeout_send_async_value_milliseconds() | % TimeoutAsync cloudi_service_api: timeout_send_sync_value_milliseconds() | % TimeoutSync cloudi_service_api: timeout_terminate_milliseconds() | % TimeoutTerm cloudi_service_api:dest_refresh() | % DestRefresh list(cloudi:service_name_pattern()) | undefined | % DestListDeny list(cloudi:service_name_pattern()) | undefined | % DestListAllow #config_service_options{} | % ConfigOptions cloudi_service_api:service_id()). % ID -type arguments_execution() :: arguments_execution_internal() | arguments_execution_external(). -export_type([arguments_execution_internal/0, arguments_execution_external/0, arguments_execution/0]). %%%------------------------------------------------------------------------ %%% External interface %%%------------------------------------------------------------------------ -spec start_internal(ProcessIndex :: non_neg_integer(), ProcessCount :: pos_integer(), TimeStart :: cloudi_timestamp:native_monotonic(), TimeRestart :: undefined | cloudi_timestamp:native_monotonic(), Restarts :: non_neg_integer(), GroupLeader :: pid(), Module :: atom(), Args :: list(), Timeout :: cloudi_service_api:timeout_initialize_milliseconds(), Prefix :: cloudi:service_name_pattern(), TimeoutAsync :: cloudi_service_api: timeout_send_async_value_milliseconds(), TimeoutSync :: cloudi_service_api: timeout_send_sync_value_milliseconds(), TimeoutTerm :: cloudi_service_api:timeout_terminate_milliseconds(), DestRefresh :: cloudi_service_api:dest_refresh(), DestListDeny :: list(cloudi:service_name_pattern()) | undefined, DestListAllow :: list(cloudi:service_name_pattern()) | undefined, ConfigOptions :: #config_service_options{}, ID :: cloudi_service_api:service_id()) -> {ok, cloudi_service:source()} | {error, error_reason_start_internal() | any()}. start_internal(ProcessIndex, ProcessCount, TimeStart, TimeRestart, Restarts, GroupLeader, Module, Args, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, DestListDeny, DestListAllow, ConfigOptions, ID) when is_integer(ProcessIndex), is_integer(ProcessCount), is_integer(TimeStart), is_integer(Restarts), is_atom(Module), is_list(Args), is_integer(Timeout), is_list(Prefix), is_integer(TimeoutAsync), is_integer(TimeoutSync), is_integer(TimeoutTerm), is_record(ConfigOptions, config_service_options), is_binary(ID) -> true = (DestRefresh =:= immediate_closest) orelse (DestRefresh =:= lazy_closest) orelse (DestRefresh =:= immediate_furthest) orelse (DestRefresh =:= lazy_furthest) orelse (DestRefresh =:= immediate_random) orelse (DestRefresh =:= lazy_random) orelse (DestRefresh =:= immediate_local) orelse (DestRefresh =:= lazy_local) orelse (DestRefresh =:= immediate_remote) orelse (DestRefresh =:= lazy_remote) orelse (DestRefresh =:= immediate_newest) orelse (DestRefresh =:= lazy_newest) orelse (DestRefresh =:= immediate_oldest) orelse (DestRefresh =:= lazy_oldest) orelse (DestRefresh =:= none), DestDeny = if DestListDeny =:= undefined -> undefined; is_list(DestListDeny) -> trie:new(DestListDeny) end, DestAllow = if DestListAllow =:= undefined -> undefined; is_list(DestListAllow) -> trie:new(DestListAllow) end, case reltool_util:is_module_loaded(Module, Timeout) of {ok, TimeoutNew} -> % Erlang application startup is asynchronous, so wait for the % module to be loaded or timeout ?PROCESS_START_INTERNAL(ProcessIndex, ProcessCount, TimeStart, TimeRestart, Restarts, GroupLeader, Module, Args, TimeoutNew, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, DestDeny, DestAllow, ConfigOptions, ID); {error, Reason} -> ?LOG_ERROR("loading ~p failed: ~tp", [Module, Reason]), {error, {service_internal_module_not_loaded, Module}} end. -spec start_external(ProcessIndex :: non_neg_integer(), ProcessCount :: pos_integer(), TimeStart :: cloudi_timestamp:native_monotonic(), TimeRestart :: undefined | cloudi_timestamp:native_monotonic(), Restarts :: non_neg_integer(), ThreadCount :: pos_integer(), Filename :: cloudi_service_api:file_path(), Arguments :: cloudi_service_api:args_external(), Environment :: cloudi_service_api:env_external(), Protocol :: 'local' | 'tcp' | 'udp', BufferSize :: pos_integer(), Timeout :: cloudi_service_api:timeout_initialize_milliseconds(), Prefix :: cloudi:service_name_pattern(), TimeoutAsync :: cloudi_service_api: timeout_send_async_value_milliseconds(), TimeoutSync :: cloudi_service_api: timeout_send_sync_value_milliseconds(), TimeoutTerm :: cloudi_service_api:timeout_terminate_milliseconds(), DestRefresh :: cloudi_service_api:dest_refresh(), DestListDeny :: list(cloudi:service_name_pattern()) | undefined, DestListAllow :: list(cloudi:service_name_pattern()) | undefined, ConfigOptions :: #config_service_options{}, ID :: cloudi_service_api:service_id()) -> {ok, nonempty_list(cloudi_service:source())} | {error, error_reason_start_external() | any()}. start_external(ProcessIndex, ProcessCount, TimeStart, TimeRestart, Restarts, ThreadCount, Filename, Arguments, Environment, Protocol, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, DestListDeny, DestListAllow, ConfigOptions, ID) -> case start_external_spawn_params(ProcessIndex, ProcessCount, ThreadCount, Filename, Arguments, Environment, Protocol, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, ConfigOptions, ID) of {ok, SpawnProcess, SpawnProtocol, SocketPath, Rlimits, Owner, Nice, CGroup, Chroot, SyscallLock, Directory, CommandLine, FilenameNew, ArgumentsNew, EnvironmentLookup, ProcessCountMax, ProcessCountMin} -> {ok, DestDeny, DestAllow} = start_external_threads_params(DestListDeny, DestListAllow), ConfigOptionsNew = ConfigOptions#config_service_options{ cgroup = CGroup}, case start_external_threads(ThreadCount, ProcessIndex, ProcessCount, TimeStart, TimeRestart, Restarts, CommandLine, Protocol, SocketPath, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, DestDeny, DestAllow, ConfigOptionsNew, ID) of {ok, Pids, Ports} -> start_external_spawn(SpawnProcess, SpawnProtocol, SocketPath, Pids, Ports, Rlimits, Owner, Nice, CGroup, Chroot, SyscallLock, Directory, ThreadCount, CommandLine, FilenameNew, ArgumentsNew, Environment, EnvironmentLookup, Protocol, BufferSize, Timeout, TimeoutTerm, ProcessIndex, ProcessCountMax, ProcessCountMin); {error, _} = Error -> Error end; {error, _} = Error -> Error end. -spec status_internal(ProcessCount :: pos_integer(), PidsOrdered :: list(pid()), arguments_execution_internal(), Status :: nonempty_list()) -> cloudi_service_api:service_status_internal(). status_internal(ProcessCount, PidsOrdered, [_GroupLeader, Module, _Args, _TimeoutInit, Prefix, _TimeoutAsync, _TimeoutSync, _TimeoutTerm, _DestRefresh, _DestListDeny, _DestListAllow, _ConfigOptions, _ID], Status) -> [{type, internal}, {prefix, Prefix}, {module, Module}, {count_process, ProcessCount}, {pids_erlang, PidsOrdered} | Status]. -spec status_external(ProcessCount :: pos_integer(), ThreadCount :: pos_integer(), OSPidsOrdered :: list(pos_integer()), PidsOrdered :: list(pid()), arguments_execution_external(), Status :: nonempty_list()) -> cloudi_service_api:service_status_external(). status_external(ProcessCount, ThreadCount, OSPidsOrdered, PidsOrdered, [_ThreadCount, Filename, _Arguments, _Environment, _Protocol, _BufferSize, _TimeoutInit, Prefix, _TimeoutAsync, _TimeoutSync, _TimeoutTerm, _DestRefresh, _DestListDeny, _DestListAllow, _ConfigOptions, _ID], Status) -> [{type, external}, {prefix, Prefix}, {file_path, Filename}, {count_process, ProcessCount}, {count_thread, ThreadCount}, {pids_os, OSPidsOrdered}, {pids_erlang, PidsOrdered} | Status]. -spec update_external(Pids :: nonempty_list(pid()), Ports :: nonempty_list(pos_integer()), arguments_execution_external()) -> ok | {error, error_reason_start_external() | any()}. update_external(Pids, Ports, [ProcessIndex, ProcessCount, ThreadCount, Filename, Arguments, Environment, Protocol, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, _DestListDeny, _DestListAllow, ConfigOptions, ID]) -> case start_external_spawn_params(ProcessIndex, ProcessCount, ThreadCount, Filename, Arguments, Environment, Protocol, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, ConfigOptions, ID) of {ok, SpawnProcess, SpawnProtocol, SocketPath, Rlimits, Owner, Nice, CGroup, Chroot, SyscallLock, Directory, CommandLine, FilenameNew, ArgumentsNew, EnvironmentLookup, ProcessCountMax, ProcessCountMin} -> case start_external_spawn(SpawnProcess, SpawnProtocol, SocketPath, Pids, Ports, Rlimits, Owner, Nice, CGroup, Chroot, SyscallLock, Directory, ThreadCount, CommandLine, FilenameNew, ArgumentsNew, Environment, EnvironmentLookup, Protocol, BufferSize, Timeout, TimeoutTerm, ProcessIndex, ProcessCountMax, ProcessCountMin) of {ok, Pids} -> _ = [Pid ! {'cloudi_service_update_state', CommandLine} || Pid <- Pids], ok; {error, _} = Error -> Error end; {error, _} = Error -> Error end. -spec update_internal_f(DestRefreshNew :: cloudi_service_api:dest_refresh() | undefined, TimeoutInitNew :: cloudi_service_api: timeout_initialize_milliseconds() | undefined, TimeoutAsyncNew :: cloudi_service_api: timeout_send_async_value_milliseconds() | undefined, TimeoutSyncNew :: cloudi_service_api: timeout_send_sync_value_milliseconds() | undefined, DestListDenyNew :: list(cloudi:service_name_pattern()) | undefined, DestListAllowNew :: list(cloudi:service_name_pattern()) | undefined, OptionsKeys :: list(atom()), ConfigOptionsNew :: #config_service_options{}, arguments_execution_internal()) -> arguments_execution_internal(). update_internal_f(DestRefreshNew, TimeoutInitNew, TimeoutAsyncNew, TimeoutSyncNew, DestListDenyNew, DestListAllowNew, OptionsKeys, ConfigOptionsNew, [GroupLeader, Module, Args, TimeoutInitOld, Prefix, TimeoutAsyncOld, TimeoutSyncOld, TimeoutTerm, DestRefreshOld, DestListDenyOld, DestListAllowOld, ConfigOptionsOld, ID]) -> [GroupLeader, Module, Args, if TimeoutInitNew =:= undefined -> TimeoutInitOld; is_integer(TimeoutInitNew) -> TimeoutInitNew end, Prefix, if TimeoutAsyncNew =:= undefined -> TimeoutAsyncOld; is_integer(TimeoutAsyncNew) -> TimeoutAsyncNew end, if TimeoutSyncNew =:= undefined -> TimeoutSyncOld; is_integer(TimeoutSyncNew) -> TimeoutSyncNew end, TimeoutTerm, if DestRefreshNew =:= undefined -> DestRefreshOld; is_atom(DestRefreshNew) -> DestRefreshNew end, if DestListDenyNew =:= invalid -> DestListDenyOld; DestListDenyNew =:= undefined; is_list(DestListDenyNew) -> DestListDenyNew end, if DestListAllowNew =:= invalid -> DestListAllowOld; DestListAllowNew =:= undefined; is_list(DestListAllowNew) -> DestListAllowNew end, cloudi_core_i_configuration:service_options_copy(OptionsKeys, ConfigOptionsOld, ConfigOptionsNew), ID]. -spec update_external_f(FilenameNew :: cloudi_service_api:file_path() | undefined, ArgumentsNew :: cloudi_service_api:args_external() | undefined, EnvironmentNew :: cloudi_service_api:env_external() | undefined, DestRefreshNew :: cloudi_service_api:dest_refresh() | undefined, TimeoutInitNew :: cloudi_service_api: timeout_initialize_milliseconds() | undefined, TimeoutAsyncNew :: cloudi_service_api: timeout_send_async_value_milliseconds() | undefined, TimeoutSyncNew :: cloudi_service_api: timeout_send_sync_value_milliseconds() | undefined, DestListDenyNew :: list(cloudi:service_name_pattern()) | undefined, DestListAllowNew :: list(cloudi:service_name_pattern()) | undefined, OptionsKeys :: list(atom()), ConfigOptionsNew :: #config_service_options{}, arguments_execution_external()) -> arguments_execution_external(). update_external_f(FilenameNew, ArgumentsNew, EnvironmentNew, DestRefreshNew, TimeoutInitNew, TimeoutAsyncNew, TimeoutSyncNew, DestListDenyNew, DestListAllowNew, OptionsKeys, ConfigOptionsNew, [ThreadCount, FilenameOld, ArgumentsOld, EnvironmentOld, Protocol, BufferSize, TimeoutInitOld, Prefix, TimeoutAsyncOld, TimeoutSyncOld, TimeoutTerm, DestRefreshOld, DestListDenyOld, DestListAllowOld, ConfigOptionsOld, ID]) -> [ThreadCount, if is_list(FilenameNew) -> FilenameNew; FilenameNew =:= undefined -> FilenameOld end, if is_list(ArgumentsNew) -> ArgumentsNew; ArgumentsNew =:= undefined -> ArgumentsOld end, if is_list(EnvironmentNew) -> EnvironmentNew; EnvironmentNew =:= undefined -> EnvironmentOld end, Protocol, BufferSize, if is_integer(TimeoutInitNew) -> TimeoutInitNew; TimeoutInitNew =:= undefined -> TimeoutInitOld end, Prefix, if is_integer(TimeoutAsyncNew) -> TimeoutAsyncNew; TimeoutAsyncNew =:= undefined -> TimeoutAsyncOld end, if is_integer(TimeoutSyncNew) -> TimeoutSyncNew; TimeoutSyncNew =:= undefined -> TimeoutSyncOld end, TimeoutTerm, if DestRefreshNew =:= undefined -> DestRefreshOld; is_atom(DestRefreshNew) -> DestRefreshNew end, if DestListDenyNew =:= invalid -> DestListDenyOld; DestListDenyNew =:= undefined; is_list(DestListDenyNew) -> DestListDenyNew end, if DestListAllowNew =:= invalid -> DestListAllowOld; DestListAllowNew =:= undefined; is_list(DestListAllowNew) -> DestListAllowNew end, cloudi_core_i_configuration:service_options_copy(OptionsKeys, ConfigOptionsOld, ConfigOptionsNew), ID]. %%%------------------------------------------------------------------------ %%% Private functions %%%------------------------------------------------------------------------ rlimits(#config_service_options{limit = L}) -> cloudi_core_i_os_process:limit_format(L). owner(#config_service_options{owner = L}, EnvironmentLookup) -> cloudi_core_i_os_process:owner_format(L, EnvironmentLookup). nice(#config_service_options{nice = Nice}) -> Nice. cgroup(#config_service_options{cgroup = L}, EnvironmentLookup) -> cloudi_core_i_os_process:cgroup_format(L, EnvironmentLookup). chroot(#config_service_options{chroot = Chroot}, EnvironmentLookup) -> cloudi_core_i_os_process:chroot_format(Chroot, EnvironmentLookup). syscall_lock(#config_service_options{syscall_lock = SyscallLock}) -> cloudi_core_i_os_process:syscall_lock_format(SyscallLock). directory(#config_service_options{directory = Directory}, EnvironmentLookup) -> cloudi_core_i_os_process:directory_format(Directory, EnvironmentLookup). count_process_limits(ProcessCount, #config_service_options{ count_process_dynamic = CountProcessDynamic}) -> case cloudi_core_i_rate_based_configuration: count_process_dynamic_limits(CountProcessDynamic) of undefined -> {ProcessCount, ProcessCount}; {_, _} = MinMax -> MinMax end. create_socket_path(TemporaryDirectory, ID) when is_binary(ID) -> Path = filename:join([TemporaryDirectory, "cloudi_socket_" ++ uuid:uuid_to_string(ID, nodash) ++ "_"]), false = filelib:is_file(Path), Path. start_external_spawn_params(ProcessIndex, ProcessCount, ThreadCount, Filename, Arguments, Environment, Protocol, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, ConfigOptions, ID) when is_integer(ProcessIndex), is_integer(ProcessCount), is_integer(ThreadCount), ThreadCount > 0, is_list(Filename), is_list(Arguments), is_list(Environment), is_integer(BufferSize), is_integer(Timeout), is_list(Prefix), is_integer(TimeoutAsync), is_integer(TimeoutSync), is_integer(TimeoutTerm), is_record(ConfigOptions, config_service_options), is_binary(ID) -> true = (Protocol =:= tcp) orelse (Protocol =:= udp) orelse (Protocol =:= local), true = (DestRefresh =:= immediate_closest) orelse (DestRefresh =:= lazy_closest) orelse (DestRefresh =:= immediate_furthest) orelse (DestRefresh =:= lazy_furthest) orelse (DestRefresh =:= immediate_random) orelse (DestRefresh =:= lazy_random) orelse (DestRefresh =:= immediate_local) orelse (DestRefresh =:= lazy_local) orelse (DestRefresh =:= immediate_remote) orelse (DestRefresh =:= lazy_remote) orelse (DestRefresh =:= immediate_newest) orelse (DestRefresh =:= lazy_newest) orelse (DestRefresh =:= immediate_oldest) orelse (DestRefresh =:= lazy_oldest) orelse (DestRefresh =:= none), TemporaryDirectory = case os:getenv("TMPDIR") of false -> "/tmp"; L -> L end, SocketPath = create_socket_path(TemporaryDirectory, ID), EnvironmentLookup = cloudi_environment:lookup(), case start_external_spawn_params_parse(Filename, Arguments, ConfigOptions, EnvironmentLookup) of {ok, CommandLine, FilenameNew, ArgumentsNew, Chroot, Directory} -> case supool:get(?OS_SPAWN_POOL) of SpawnProcess when is_pid(SpawnProcess) -> SpawnProtocol = if Protocol =:= tcp -> $t; % inet Protocol =:= udp -> $u; % inet Protocol =:= local -> $l % tcp local (unix domain socket) end, Rlimits = rlimits(ConfigOptions), Owner = owner(ConfigOptions, EnvironmentLookup), Nice = nice(ConfigOptions), CGroup = cgroup(ConfigOptions, EnvironmentLookup), SyscallLock = syscall_lock(ConfigOptions), {ProcessCountMin, ProcessCountMax} = count_process_limits(ProcessCount, ConfigOptions), {ok, SpawnProcess, SpawnProtocol, SocketPath, Rlimits, Owner, Nice, CGroup, Chroot, SyscallLock, Directory, CommandLine, FilenameNew, ArgumentsNew, EnvironmentLookup, ProcessCountMax, ProcessCountMin}; undefined -> {error, noproc} end; {error, _} = Error -> Error end. start_external_threads_params(DestListDeny, DestListAllow) -> DestDeny = if DestListDeny =:= undefined -> undefined; is_list(DestListDeny) -> trie:new(DestListDeny) end, DestAllow = if DestListAllow =:= undefined -> undefined; is_list(DestListAllow) -> trie:new(DestListAllow) end, {ok, DestDeny, DestAllow}. start_external_spawn_params_parse(Filename, Arguments, ConfigOptions, EnvironmentLookup) -> case filename_parse(Filename, EnvironmentLookup) of {ok, FilenameNew} -> case arguments_parse(Arguments, EnvironmentLookup) of {ok, ArgumentsNew, ArgumentsList} -> CommandLine = [FilenameNew | ArgumentsList], case chroot(ConfigOptions, EnvironmentLookup) of {ok, Chroot} -> case directory(ConfigOptions, EnvironmentLookup) of {ok, Directory} -> {ok, CommandLine, FilenameNew, ArgumentsNew, Chroot, Directory}; {error, _} = Error -> Error end; {error, _} = Error -> Error end; {error, _} = Error -> Error end; {error, _} = Error -> Error end. start_external_thread(ThreadCount, Pids, Ports, ThreadCount, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _) -> {ok, lists:reverse(Pids), lists:reverse(Ports)}; start_external_thread(I, Pids, Ports, ThreadCount, ProcessIndex, ProcessCount, TimeStart, TimeRestart, Restarts, CommandLine, Protocol, SocketPath, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, DestDeny, DestAllow, #config_service_options{ bind = Bind} = ConfigOptions, ID) -> case ?PROCESS_START_EXTERNAL(Protocol, SocketPath, I + ThreadCount * ProcessIndex, ProcessIndex, ProcessCount, TimeStart, TimeRestart, Restarts, CommandLine, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, DestDeny, DestAllow, ConfigOptions, ID) of {ok, Pid, Port} -> BindNew = cloudi_core_i_concurrency:bind_increment_thread(Bind), start_external_thread(I + 1, [Pid | Pids], [Port | Ports], ThreadCount, ProcessIndex, ProcessCount, TimeStart, TimeRestart, Restarts, CommandLine, Protocol, SocketPath, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, DestDeny, DestAllow, ConfigOptions#config_service_options{ bind = BindNew}, ID); {error, _} = Error -> lists:foreach(fun(P) -> erlang:exit(P, kill) end, Pids), Error end. start_external_threads(ThreadCount, ProcessIndex, ProcessCount, TimeStart, TimeRestart, Restarts, CommandLine, Protocol, SocketPath, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, DestDeny, DestAllow, ConfigOptions, ID) -> start_external_thread(0, [], [], ThreadCount, ProcessIndex, ProcessCount, TimeStart, TimeRestart, Restarts, CommandLine, Protocol, SocketPath, BufferSize, Timeout, Prefix, TimeoutAsync, TimeoutSync, TimeoutTerm, DestRefresh, DestDeny, DestAllow, ConfigOptions, ID). start_external_spawn(SpawnProcess, SpawnProtocol, SocketPath, Pids, Ports, Rlimits, Owner, Nice, CGroup, Chroot, SyscallLock, Directory, ThreadCount, CommandLine, Filename, Arguments, Environment, EnvironmentLookup, Protocol, BufferSize, Timeout, TimeoutTerm, ProcessIndex, ProcessCountMax, ProcessCountMin) -> case environment_parse(Environment, ThreadCount, Protocol, BufferSize, Timeout, TimeoutTerm, ProcessIndex, ProcessCountMax, ProcessCountMin, EnvironmentLookup) of {ok, SpawnEnvironment} -> {UserI, UserStr, GroupI, GroupStr} = Owner, case cloudi_core_i_os_spawn:spawn(SpawnProcess, SpawnProtocol, string_terminate(SocketPath), Ports, Rlimits, UserI, string_terminate(UserStr), GroupI, string_terminate(GroupStr), Nice, string_terminate(Chroot), SyscallLock, string_terminate(Directory), string_terminate(Filename), string_terminate(Arguments), SpawnEnvironment) of {ok, OSPid} -> ?LOG_INFO("OS pid ~p spawned ~p~n ~tp", [OSPid, Pids, CommandLine]), case cloudi_core_i_os_process:cgroup_set(OSPid, CGroup) of ok -> {ok, Pids}; {error, _} = Error -> Error end; {error, _} = Error -> Error end; {error, _} = Error -> Error end. % update filename, including path filename_parse(Filename, EnvironmentLookup) -> case cloudi_environment:transform(Filename, EnvironmentLookup) of [] -> {error, {service_external_file_path_invalid_expanded, Filename}}; ExpandedFilename -> case unicode:characters_to_binary(ExpandedFilename, utf8) of {_, _, _} -> {error, {service_external_file_path_invalid_unicode, ExpandedFilename}}; Utf8Filename -> {ok, erlang:binary_to_list(Utf8Filename)} end end. % remove beginning whitespace and validate delimiters % within the command-line arguments arguments_parse([32 | Arguments], EnvironmentLookup) -> arguments_parse(Arguments, EnvironmentLookup); arguments_parse(Arguments, EnvironmentLookup) -> ArgumentsNew = cloudi_environment:transform(Arguments, EnvironmentLookup), case arguments_parse([], [], [], none, ArgumentsNew) of {ok, _, _} = Success -> Success; Reason when is_atom(Reason) -> {error, {Reason, Arguments}} end. arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, []) -> {ok, lists:reverse(ArgumentsBinary), lists:reverse([lists:reverse(Argument) | ArgumentsList])}; arguments_parse(_, _, _, _, []) -> service_external_args_invalid_quotes; arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, [$' | T]) -> arguments_parse(ArgumentsBinary, ArgumentsList, Argument, $', T); arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, [$" | T]) -> arguments_parse(ArgumentsBinary, ArgumentsList, Argument, $", T); arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, [$` | T]) -> arguments_parse(ArgumentsBinary, ArgumentsList, Argument, $`, T); arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, [32 | [32 | _] = T]) -> arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, T); arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, [32 | T]) -> arguments_parse([0 | ArgumentsBinary], [lists:reverse(Argument) | ArgumentsList], [], none, T); arguments_parse(ArgumentsBinary, ArgumentsList, Argument, $', [$' | T]) -> arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, T); arguments_parse(ArgumentsBinary, ArgumentsList, Argument, $", [$" | T]) -> arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, T); arguments_parse(ArgumentsBinary, ArgumentsList, Argument, $`, [$` | T]) -> arguments_parse(ArgumentsBinary, ArgumentsList, Argument, none, T); arguments_parse(ArgumentsBinary, ArgumentsList, Argument, Delim, [H | T]) -> case unicode:characters_to_binary([H], utf8) of {_, _, _} -> service_external_args_invalid_unicode; Utf8H -> ArgumentsBinaryNew = lists:reverse(erlang:binary_to_list(Utf8H), ArgumentsBinary), arguments_parse(ArgumentsBinaryNew, ArgumentsList, [H | Argument], Delim, T) end. % add CloudI API environmental variables and format into a single % string that is easy to use in C/C++ environment_parse(Environment0, ThreadCount0, Protocol0, BufferSize0, Timeout0, TimeoutTerm0, ProcessIndex0, ProcessCountMax0, ProcessCountMin0, EnvironmentLookup0) -> ThreadCount1 = erlang:integer_to_list(ThreadCount0), Protocol1 = erlang:atom_to_list(Protocol0), BufferSize1 = erlang:integer_to_list(BufferSize0), Timeout1 = erlang:integer_to_list(Timeout0), TimeoutTerm1 = erlang:integer_to_list(TimeoutTerm0), ProcessIndex1 = erlang:integer_to_list(ProcessIndex0), ProcessCountMax1 = erlang:integer_to_list(ProcessCountMax0), ProcessCountMin1 = erlang:integer_to_list(ProcessCountMin0), Environment1 = lists:keystore(?ENVIRONMENT_THREAD_COUNT, 1, Environment0, {?ENVIRONMENT_THREAD_COUNT, ThreadCount1}), Environment2 = lists:keystore(?ENVIRONMENT_PROTOCOL, 1, Environment1, {?ENVIRONMENT_PROTOCOL, Protocol1}), Environment3 = lists:keystore(?ENVIRONMENT_BUFFER_SIZE, 1, Environment2, {?ENVIRONMENT_BUFFER_SIZE, BufferSize1}), Environment4 = lists:keystore(?ENVIRONMENT_TIMEOUT_INITIALIZE, 1, Environment3, {?ENVIRONMENT_TIMEOUT_INITIALIZE, Timeout1}), Environment5 = lists:keystore(?ENVIRONMENT_TIMEOUT_TERMINATE, 1, Environment4, {?ENVIRONMENT_TIMEOUT_TERMINATE, TimeoutTerm1}), Environment6 = lists:keystore(?ENVIRONMENT_PROCESS_INDEX, 1, Environment5, {?ENVIRONMENT_PROCESS_INDEX, ProcessIndex1}), Environment7 = lists:keystore(?ENVIRONMENT_PROCESS_COUNT_MAX, 1, Environment6, {?ENVIRONMENT_PROCESS_COUNT_MAX, ProcessCountMax1}), Environment8 = lists:keystore(?ENVIRONMENT_PROCESS_COUNT_MIN, 1, Environment7, {?ENVIRONMENT_PROCESS_COUNT_MIN, ProcessCountMin1}), EnvironmentLookup1 = trie:store(?ENVIRONMENT_THREAD_COUNT, ThreadCount1, EnvironmentLookup0), EnvironmentLookup2 = trie:store(?ENVIRONMENT_PROTOCOL, Protocol1, EnvironmentLookup1), EnvironmentLookup3 = trie:store(?ENVIRONMENT_BUFFER_SIZE, BufferSize1, EnvironmentLookup2), EnvironmentLookup4 = trie:store(?ENVIRONMENT_TIMEOUT_INITIALIZE, Timeout1, EnvironmentLookup3), EnvironmentLookup5 = trie:store(?ENVIRONMENT_TIMEOUT_TERMINATE, TimeoutTerm1, EnvironmentLookup4), EnvironmentLookup6 = trie:store(?ENVIRONMENT_PROCESS_INDEX, ProcessIndex1, EnvironmentLookup5), EnvironmentLookup7 = trie:store(?ENVIRONMENT_PROCESS_COUNT_MAX, ProcessCountMax1, EnvironmentLookup6), EnvironmentLookup8 = trie:store(?ENVIRONMENT_PROCESS_COUNT_MIN, ProcessCountMin1, EnvironmentLookup7), environment_format(Environment8, EnvironmentLookup8). environment_format_value([], _, K, _) -> {error, {service_external_env_invalid_expanded, K}}; environment_format_value(_, [], _, _) -> {ok, []}; environment_format_value([_ | _] = ExpandedK, [_ | _] = ExpandedV, K, V) -> case unicode:characters_to_binary(ExpandedK, utf8) of {_, _, _} -> {error, {service_external_env_invalid_unicode, K}}; Utf8K -> case unicode:characters_to_binary(ExpandedV, utf8) of {_, _, _} -> {error, {service_external_env_invalid_unicode, V}}; Utf8V -> {ok, [Utf8K, $=, Utf8V, 0]} end end. environment_format(Environment, EnvironmentLookup) -> environment_format([], Environment, EnvironmentLookup). environment_format(Output, [], _) -> {ok, erlang:iolist_to_binary(lists:reverse(Output))}; environment_format(Output, [{K, V} | Environment], EnvironmentLookup) -> ExpandedK = cloudi_environment:transform(K, EnvironmentLookup), ExpandedV = cloudi_environment:transform(V, EnvironmentLookup), case environment_format_value(ExpandedK, ExpandedV, K, V) of {ok, Pair} -> environment_format([Pair | Output], Environment, EnvironmentLookup); {error, _} = Error -> Error end. % terminate the string for easy access within C/C++ string_terminate([]) -> <<0>>; string_terminate([_ | _] = L) -> erlang:iolist_to_binary([L, 0]).