MqMsgque PACKAGE - the L)anguage A)bstraction L)ayer definition … More...

Collaboration diagram for MqMsgque_Sys_C_API:

Data Structures
struct	MqSysServerThreadMainS
	data used to initialize a new created thread … More...

struct	MqLalS
	Interface between MqContextS and the Operating-System … More...

Typedefs
typedef int(*	MqSysSelectF) (int max, void read, void write, void except, struct timeval timeout)
	select syscall …

typedef enum MkErrorE(*	MqSysServerSpawnF) (MK_RT mkrt, MQ_RT mqrt, MQ_CTX const ctx, char const argv, MK_STRN name, struct MkIdS *idPtrOut)
	spawn server create syscall with libmqmsgque error plugin

typedef enum MkErrorE(*	MqSysServerThreadF) (MK_RT mkrt, MQ_CTX const ctx, struct MqSysServerThreadMainS const argP, MK_STRN name, int state, struct MkIdS idPtrOut)
	thread server create syscall with libmqmsgque error plugin

typedef enum MkErrorE(*	MqSysServerForkF) (MK_RT mkrt, MQ_RT mqrt, MQ_CTX const ctx, MQ_FCT factory, MK_BFL argsP, MK_STRN name, MQ_SOCK_HDL socket, struct MkIdS idPtrOut)
	fork server create syscall with libmqmsgque error plugin

Functions
void	MqSysServerThreadMain (struct MqSysServerThreadMainS *data)
	initialize a new thread …

Variables
struct MqLalS	MqLal
	Language Abstraction Layer in duty.

Detailed Description

MqMsgque PACKAGE - the L)anguage A)bstraction L)ayer definition …

The LAL is required to give the target of embedding a change to modify the access to system functions. RUBY for example has it's own Thread-Creation function.

All wrappers to access the underlying OS,Languages and system functions are located in this section. Every external Function has additional code to map the OS/System specific error codes/messages into a libmqmsgque specific error codes/messages

Typedef Documentation

◆ MqSysSelectF

typedef int(* MqSysSelectF) (int max, void *read, void *write, void *except, struct timeval *timeout)

select syscall …

◆ MqSysServerForkF

typedef enum MkErrorE(* MqSysServerForkF) (MK_RT mkrt, MQ_RT mqrt, MQ_CTX const ctx, MQ_FCT factory, MK_BFL *argsP, MK_STRN name, MQ_SOCK_HDL socket, struct MkIdS *idPtrOut)

fork server create syscall with libmqmsgque error plugin

Parameters

[in]	mkrt	the MkRuntimeS instance to work on - the runtime argument, used by MK_RT_CALL (C-only)
[in]	mqrt	the MqRuntimeS instance to work on - the runtime argument, used by MQ_RT_CALL (C-only)
[in]	ctx	the MqContextS instance to work on
[in,out]	factory	server configuration (memory will be freed)
[in]	argsP	command-line arguments before MK_ALFA, owned by SysServerThread
[in]	name	the name of the thread
[in]	socket	the name of the socket the ne FORK have to listen
[out]	idPtrOut	the process identifier, pointer is not `MK_NULL`

Returns: The MkErrorE status from the DEFAULT MkErrorC in MkRuntimeS::error_mk

Example:

static enum MkErrorE sSysServerFork (
  MQ_RT_ARGS
  MQ_CTX const context,           
  struct MqFactoryS* factory,     
  struct MkBufferListS ** argvP,  
  MK_STRN  name,                  
  MQ_SOCK_HDL socket,             
  struct MkIdS * idP              
) {
#if defined(HAVE_FORK)
//M0
//printSTACK_0();
  
  // fork setup !before! a fork is created
  if (MqContextCT_X(context)->fParentBeforeFork) {
    (*MqContextCT_X(context)->fParentBeforeFork)(context);
  }
 
//printULSS1(*argvP);
 
  // this is used for a filter pipeline like "| atool split .. @ cut ... @ join ..."
  // argv[0] is set by the tool (cut or join) and need !not! be set by name
  MkErrorCheck(pMkSysFork(MK_RT_CALL MkOBJ(context),idP));
  if ((*idP).val == 0UL) {
    // prevent the "client-context" from deleting in the new process
    // reason: perl has a garbage collection, after fork this "context" is in the
    // new process but with an invalid "event-link" (pIoCreate was done but pIoconnect not)
    // and "ppMqEventCreate" can not set the "MqContextDelete_LOCK"
    context->bits.MqContextDelete_LOCK = true;
    // cleanup "events" after FORK
    MkErrorCheck(pEventAfterFork(MQ_RT_CALL context));
    // child AFTER fork
    if (MqContextCT_X(context)->fChildAfterFork) (*MqContextCT_X(context)->fChildAfterFork)(context);
    // setup new context
    MQ_CTX newctx = NULL;
    // create the new context
//M1
//MqFactoryLog_1(factory);
    check_MkErrorE(MqFactoryInvoke (context, MQ_FACTORY_NEW_FORK, factory, &newctx)) goto error1;
//M2
//printC(MqClassIdentGet(newctx))
    // add my configuration
    newctx->statusIs = (enum MqStatusIsEF) (newctx->statusIs | MQ_STATUS_IS_FORK);
    // the new CONTEXT is always a SERVER
    newctx->setup.isServer = true;
    // BUG-FIX: worker-2-1-0-
    // DESCR: display-lock inherited from "client" disable prefix-name-setting on "server"
    //        -> "spawn" does NOT inherited locking at all
    newctx->config.bits.dispNameLOCK = false;
    newctx->config.bits.storageFileLOCK = false;
    newctx->config.bits.dispPrefixLOCK= false;
    newctx->config.bits.dispPostfixLOCK= false;
 
    // BUG-FIX:   error-6-ERR8-1-
    // TEST:      Nhi1Exec error.test --lang-ruby --block-6 -match error-6-ERR8-1-*
    // ANALYSE:   On FORK the refCount management (MK_RT_REF.cid, MK_RT_REF.cidA, MK_RT_REF.cidN) is already used by
    //            CALLING context (client). On RUBY the EXIT from the FORK (server) process does NOT intermedialtly stop
    //            because the rb_exit is used. This rb_exit do some RUBY cleanup and later on step back to the invalid 
    //            CALLING (client!!) code in the FORKED (server) process. this code has now an UNBALENCED MK_RT_REF.cid
    //            and create and error.
    // FIX:       delete the calling context from the MK_RT_REF.cid
    // UPDATE:    this crash C… and propably more. PROBLEN is now RUBY→rb_exit
    //pRefDecrO(MK_RT_CALL MkObj(MK_RT_REF.cid), __func__);
 
    // configure the link
    // 1.   ---duplicate
    pConfigSetDuplicate (newctx);
    // 2.   ---pipe-socket HDL
    MqConfigSetIoPipe (newctx, socket);
    // 3.   --thread
    MqConfigSetStartAs_E (newctx, MQ_START_FORK);
 
    // create link
    check_MkErrorE (MqLinkCreate(newctx,*argvP)) goto error1;
 
    // the MqProcessEvent is necessary because "ParentCreate" have to come back
    // ifNot: java create object will fail
    if (newctx->setup.isServer == true) {
      MqProcessEvent(newctx, MQ_WAIT_FOREVER, MK_TIMEOUT_USER);
    }
error1:
    MqExit_1(newctx);
  } else {
    // parend AFTER fork
    if (MqContextCT_X(context)->fParentAfterFork) (*MqContextCT_X(context)->fParentAfterFork)(context);
  }
  pBflDelete_NULL((*argvP));
  // do not create a "defunc" process
  // this is "client" code… no "IgnorSIGCHLD" in client
  //MkErrorCheck (MkSysIgnorSIGCHLD(context));
  return MK_OK;
 
error:
  return MkErrorDbV_1_XS (MK_ERROR_CAN_NOT_START_SERVER, name);
#else
  return MkErrorDbV_1_XS (MK_ERROR_NOT_SUPPORTED, "sSysServerFork");
#endif   /* HAVE_FORK */
}

Definition at line 2709 of file LibMqMsgque_mq.h.

◆ MqSysServerSpawnF

typedef enum MkErrorE(* MqSysServerSpawnF) (MK_RT mkrt, MQ_RT mqrt, MQ_CTX const ctx, char *const *argv, MK_STRN name, struct MkIdS *idPtrOut)

spawn server create syscall with libmqmsgque error plugin

Parameters

[in]	mkrt	the MkRuntimeS instance to work on - the runtime argument, used by MK_RT_CALL (C-only)
[in]	mqrt	the MqRuntimeS instance to work on - the runtime argument, used by MQ_RT_CALL (C-only)
[in]	ctx	the MqContextS instance to work on
[in]	argv	command-line arguments
[in]	name	the name of the process
[out]	idPtrOut	the process identifier, pointer is not `MK_NULL`

Returns: The MkErrorE status from the DEFAULT MkErrorC in MkRuntimeS::error_mk

Example:

static enum MkErrorE sSysServerSpawn (
  MQ_RT_ARGS
  MQ_CTX const context,
  char *const *argv,
  MK_STRN  name,
  struct MkIdS * idP
) {
  pid_t pid;
  MK_STRN cmd="sSysServerSpawn";
  int err=0;
 
  // spawn setup !before! a spawn is created
  if (MqContextCT_X(context)->fSpawnInit) (*MqContextCT_X(context)->fSpawnInit)(context);
 
#if defined(META_IS_POSIX)
 
  // get the PATH of "name"… search the "PATH" environment variable
  char path[PATH_MAX+1];
  if (findpathof(path,name) == 0) {
    cmd="findpathof";
    goto error;
  }
 
  // perl !need! sigschiled enabled at startup make it the default
  #if defined(HAVE_POSIX_SPAWN)
    {
      posix_spawnattr_t sa;
      if ((err=posix_spawnattr_init(&sa)) != 0) {
        cmd="posix_spawnattr_init";
        goto error;
      }
 
      // problem: posix_spawnp (using "name" as parameter) comes back with SUCCESS even if "name" has bogo data
      // solution: check "name" by myselv (findpathof) and use the "non-P" version…
      if (unlikely ((err=posix_spawn(&pid, path, NULL, &sa, (char *const *) argv, environ)) != 0)) {
        posix_spawnattr_destroy(&sa);
        cmd="posix_spawnp";
        goto error;
      }
 
      if ((err=posix_spawnattr_destroy(&sa)) != 0) {
        cmd="posix_spawnattr_destroy";
        goto error;
      }
 
    }
  goto ok;
 
  //  ./process.h:int spawnl(int mode, const char *path, const char *argv0, ...);
  //  ./process.h:int spawnle(int mode, const char *path, const char *argv0, ... /*, char * const *envp */);
  //  ./process.h:int spawnlp(int mode, const char *path, const char *argv0, ...);
  //  ./process.h:int spawnlpe(int mode, const char *path, const char *argv0, ... /*, char * const *envp */);
  //  ./process.h:int spawnv(int mode, const char *path, const char * const *argv);
  //  ./process.h:int spawnve(int mode, const char *path, const char * const *argv, const char * const *envp);
  //  ./process.h:int spawnvp(int mode, const char *path, const char * const *argv);
  //  ./process.h:int spawnvpe(int mode, const char *path, const char * const *argv, const char * const *envp);
 
 
  #elif (defined(HAVE_FORK) || defined(HAVE_VFORK)) && defined(HAVE_EXECV)
 
    // fork to create the child
  # if defined(HAVE_VFORK)
    if (unlikely ((pid = vfork()) == -1)) {
      cmd="vfork";
      err=errno;
      goto error;
    }
  # elif defined(HAVE_FORK)
    if (unlikely ((pid = MqSysFork()) == -1)) {
      cmd="MqSysFork";
      err=errno;
      goto error;
    }
  # else
    goto error;
  # endif
    if (pid == 0) {
      // this is the child
      if (execv(path, argv) == -1) {
        _exit (EXIT_FAILURE);
      }
      // the child will never reach this
    }
    goto ok;
 
  #else
 
  # error "unable to spwawn a process"
 
  #endif /* HAVE_POSIX_SPAWN, HAVE_FORK, HAVE_VFORK, HAVE_EXECV */
 
#else /* META_IS_WINDOWS */
  {
    char buf[2048];
    char *nbuf=buf;
 
    // build command-line, windows _spawnvp is NOT able to handle !! whitespace !!
    // in path name argument
    for (;*argv != NULL; argv++) {
      nbuf += sprintf(nbuf, "\"%s\" ", *argv);
    }
 
    // start process
    // java on windows test-cases using tcltest seems not be able to spawn new processes
    // unknown reason. command line works: 
    // >NhiExec example.Server.java --tcp --port 7777 --spawn
    errno=0;
    if (unlikely ((pid = _spawnlp (_P_NOWAIT, name, buf, NULL)) == -1)) {
      cmd="_spawnvp";
      err = errno;
      goto error;
    }
    goto ok;
  }
#endif
 
ok:
  (*idP).val = (MK_IDNT)pid;
  (*idP).type = MK_ID_PROCESS;
  // if NO child process is available an: "POSIX-no children" error is set
  // this is the "client" site an a "MkSysIgnorSIGCHLD" will interact with 
  // a "programming-language" like TCL… → avoid it
  //MkSysIgnorSIGCHLD(MK_ERROR_IGNORE);
  return MK_OK;
 
error:
  MkErrorDbV_1_XS (MK_ERROR_CAN_NOT_START_SERVER, name);
  if (err != 0) {
    MkErrorAppendV_2M(context, "can not '%s' -> [%i] ERR<%s>", cmd, err, strerror (err));
  }

Definition at line 2709 of file LibMqMsgque_mq.h.

◆ MqSysServerThreadF

typedef enum MkErrorE(* MqSysServerThreadF) (MK_RT mkrt, MQ_CTX const ctx, struct MqSysServerThreadMainS *const argP, MK_STRN name, int state, struct MkIdS *idPtrOut)

thread server create syscall with libmqmsgque error plugin

Parameters

[in]	mkrt	the MkRuntimeS instance to work on - the runtime argument, used by MK_RT_CALL (C-only)
[in]	ctx	the MqContextS instance to work on
[in]	argP	thread startup environment
[in]	name	the name of the thread
[in]	state	state of the thread `PTHREAD_CREATE_DETACHED` or `PTHREAD_CREATE_JOINABLE`
[out]	idPtrOut	the thread identifier, pointer is not `MK_NULL`

Returns: The MkErrorE status from the DEFAULT MkErrorC in MkRuntimeS::error_mk

Example:

static enum MkErrorE sSysServerThread (
  MK_RT_ARGS
  MQ_CTX const context,
  struct MqSysServerThreadMainS * const argP,
  MK_STRN  name,                
  int thread_status,            
  struct MkIdS * idP    
) {
#if META_HAS_THREAD
  mqthread_t threadId;
 
  // after a "thread" no "fork" is possible
  MqContextCT_X(context)->ignoreFork = true;
 
  # if defined(HAVE_PTHREAD)
 
  #   define check0(cmd,...) if (cmd(__VA_ARGS__)!=0) {MkErrorSetC_1XS("Error in '" #cmd "'");goto error;}
 
    int ret;
 
    // thread attributes
    pthread_attr_t attr;
    check0(pthread_attr_init, &attr);
/*
      printXV(context,"thread_status<%s>",
        thread_status==PTHREAD_CREATE_DETACHED?"PTHREAD_CREATE_DETACHED":"PTHREAD_CREATE_JOINABLE");
*/
    check0(pthread_attr_setdetachstate, &attr, thread_status);
    check0(pthread_attr_setscope, &attr, PTHREAD_SCOPE_SYSTEM);
 
    // thread setup !before! a thread was created
    if (MqContextCT_X(context)->fThreadInit) (*MqContextCT_X(context)->fThreadInit)(context);
 
    // start thread
    do {
      ret = pthread_create(&threadId,&attr,sSysServerThreadInit,argP);
    } while (ret == EAGAIN);
 
    // cleanup attribute
    check0(pthread_attr_destroy, &attr);
 
    if (unlikely(ret != 0)) {
      MkErrorDbV_1_XS (MK_ERROR_CAN_NOT_START_SERVER, name);
      goto error;
    }
 
  # else    // NOT defined(HAVE_PTHREAD)
 
    if (unlikely ( (threadId = _beginthreadex(NULL, 0, sSysServerThreadInit, argP, 0, NULL)) == 0)) {
      MkErrorDbV_1_XS (MK_ERROR_CAN_NOT_START_SERVER, name);
      MkErrorSysAppend (_beginthreadex, errno);
      goto error;
    }
 
  # endif   // defined(HAVE_PTHREAD)
 
  // save tid
  (*idP).val = (MK_IDNT)threadId;
  (*idP).type = MK_ID_THREAD;
 
  return MK_OK;
 
error:
  pBufferStreamFree(MK_RT_CALL MkBUS(&argP->argvR));
  #if MqRuntimeC_SysServerThread_size
    argP->isUsed = false;
  #else
    MkSysFreeNonNull(argP);
  #endif
  return MK_ERROR;
#else     // NOT META_HAS_THREAD
  return MkErrorDbV_1_XS (MK_ERROR_NOT_SUPPORTED, "sSysServerThread");
#endif    // META_HAS_THREAD
}

Definition at line 2709 of file LibMqMsgque_mq.h.

Function Documentation

◆ MqSysServerThreadMain()

void MqSysServerThreadMain ( struct MqSysServerThreadMainS * data )

initialize a new thread …

A libmqmsgque thread can be created with the MqLalS::MqSysServerThreadCB "included" or by a new target-language-specific MqLalS::MqSysServerThreadCB. If a target-language-specific MqLalS::MqSysServerThreadCB is used, it is required to call the LibMqMsgque MqSysServerThreadMain to initialize the new thread proper.

pthread_create(&threadId,&attr,sSysServerThreadInit,argP);
...
// wrapper to hide "MqSysServerThreadMain" from "pthread_create"
static mqthread_ret_t mqthread_stdcall sSysServerThreadInit (
 void * data
)
{
  // do the "target-language-specific" initialization,
  // can be used to set "MqSysServerThreadMainS::mkThreadData" etc...
  MqSysServerThreadMain((struct MqSysServerThreadMainS*)data);
  return mqthread_ret_NULL;
}
...

Parameters

[in] data configuration data for the new thread

Variable Documentation

◆ MqLal

struct MqLalS MqLal

extern

Language Abstraction Layer in duty.

[in]	max	the maximum file descriptor + 1 from the read, write or except input data
[in]	read	set of file descriptors
[in]	write	set of file descriptors
[in]	except	set of file descriptors
[in]	timeout	maximum time to wait for data

Data Structures

Typedefs

Functions

Variables

Detailed Description

Typedef Documentation

◆ MqSysSelectF

◆ MqSysServerForkF

◆ MqSysServerSpawnF

◆ MqSysServerThreadF

Function Documentation

◆ MqSysServerThreadMain()

Variable Documentation

◆ MqLal