--- libaitio/src/exec.c 2013/12/05 13:22:10 1.1.2.3 +++ libaitio/src/exec.c 2013/12/05 23:43:46 1.1.2.10 @@ -28,17 +28,28 @@ io_progInit(const char *progName, u_int initNum, u_int prg->prog_maxn = maxNum; strlcpy(prg->prog_name, progName, sizeof prg->prog_name); + prg->prog_used = e_malloc(E_ALIGN(prg->prog_maxn, sizeof *prg->prog_used) / + sizeof *prg->prog_used); + if (!prg->prog_used) { + io_SetErr(elwix_GetErrno(), "%s", elwix_GetError()); + e_free(prg); + return NULL; + } + prg->prog_fds = array_Init(prg->prog_maxn); if (!prg->prog_fds) { io_SetErr(elwix_GetErrno(), "%s", elwix_GetError()); + e_free(prg->prog_used); e_free(prg); return NULL; } pthread_mutex_init(&prg->prog_mtx, NULL); - if (io_progOpen(prg, prg->prog_inin) < 0) + if (io_progOpen(prg, prg->prog_inin) < 0) { io_progDestroy(&prg); + prg = NULL; + } return prg; } @@ -56,6 +67,7 @@ io_progDestroy(prog_t ** __restrict pprg) io_progClose(*pprg, 0); + e_free((*pprg)->prog_used); array_Destroy(&(*pprg)->prog_fds); pthread_mutex_destroy(&(*pprg)->prog_mtx); @@ -87,8 +99,9 @@ io_progClose(prog_t * __restrict prg, u_int closeNum) for (i = array_Size(prg->prog_fds) - 1; (closeNum ? ret < closeNum : 42) && i > -1; i--) if (array_Get(prg->prog_fds, i)) { - pclose(array(prg->prog_fds, i, FILE*)); + e_pclose(array(prg->prog_fds, i, FILE*)); array_Del(prg->prog_fds, i, 0); + clrbit(prg->prog_used, i); prg->prog_cnum--; ret++; } @@ -108,12 +121,13 @@ int io_progOpen(prog_t * __restrict prg, u_int execNum) { FILE *f; - int ret = 0; + int stat, ret = 0; register int i; + pid_t pid; if (!prg) return 0; - if (execNum > prg->prog_maxn) { + if (prg->prog_cnum + execNum > prg->prog_maxn) { io_SetErr(EINVAL, "Requested number for program execution is over pool's limit"); return 0; } @@ -121,11 +135,16 @@ io_progOpen(prog_t * __restrict prg, u_int execNum) pthread_mutex_lock(&prg->prog_mtx); for (i = 0; (execNum ? ret < execNum : 42) && i < array_Size(prg->prog_fds); i++) if (!array_Get(prg->prog_fds, i)) { - f = popen(prg->prog_name, "r+"); + f = e_popen(prg->prog_name, "r+", &pid); if (!f) { LOGERR; - ret *= -1; + ret = -1; break; + } else if (waitpid(pid, &stat, WNOHANG)) { + io_SetErr(ECHILD, "Program with pid=%d exit with status %d", + pid, WIFEXITED(stat) ? WEXITSTATUS(stat) : -1); + ret = -1; + break; } else array_Set(prg->prog_fds, i, f); prg->prog_cnum++; @@ -137,6 +156,28 @@ io_progOpen(prog_t * __restrict prg, u_int execNum) } /* + * io_progGrow() - Execute to number of programs in pool + * + * @prg = program pool + * @toNum = execute to number of programs (0 max) + * return: 0 error, >0 executed programs and abs(<0) executed programs with logged error + */ +int +io_progGrow(prog_t * __restrict prg, u_int toNum) +{ + if (!prg) + return 0; + if (toNum > prg->prog_maxn) { + io_SetErr(EINVAL, "Requested number for program execution is over pool's limit"); + return 0; + } + if (!toNum) + toNum = prg->prog_maxn; + + return io_progOpen(prg, toNum - prg->prog_cnum); +} + +/* * io_progVacuum() - Vacuum pool to running number of programs * * @prg = program pool @@ -160,8 +201,8 @@ io_progVacuum(prog_t * __restrict prg, u_int toNum) pthread_mutex_lock(&prg->prog_mtx); for (i = array_Size(prg->prog_fds) - 1; prg->prog_cnum > toNum && i > -1; i--) - if (array_Get(prg->prog_fds, i)) { - pclose(array(prg->prog_fds, i, FILE*)); + if (array_Get(prg->prog_fds, i) && isclr(prg->prog_used, i)) { + e_pclose(array(prg->prog_fds, i, FILE*)); array_Del(prg->prog_fds, i, 0); prg->prog_cnum--; ret++; @@ -169,4 +210,84 @@ io_progVacuum(prog_t * __restrict prg, u_int toNum) pthread_mutex_unlock(&prg->prog_mtx); return ret; +} + +/* + * io_progCheck() - Check exit status of program pool + * + * @prg = program pool + * return: -1 error or >-1 exited programs + */ +int +io_progCheck(prog_t * __restrict prg) +{ + int ret = 0; + struct tagPIOPID *p; + register int i; + + if (!prg) + return -1; + + pthread_mutex_lock(&prg->prog_mtx); + for (i = 0; i < array_Size(prg->prog_fds); i++) + if (array_Get(prg->prog_fds, i) && + (p = pio_pgetpid(array(prg->prog_fds, i, FILE*)))) + if (waitpid(p->pid, &p->stat, WNOHANG) > 0) { + clrbit(prg->prog_used, i); + ret++; + } + pthread_mutex_unlock(&prg->prog_mtx); + + return ret; +} + +/* + * io_progAttach() - Attach to open program + * + * @prg = program pool + * return: NULL error or !=NULL attached program handle + */ +FILE * +io_progAttach(prog_t * __restrict prg) +{ + FILE *f = NULL; + register int i; + + if (!prg) + return NULL; + + pthread_mutex_lock(&prg->prog_mtx); + for (i = 0; i < array_Size(prg->prog_fds); i++) + if (array_Get(prg->prog_fds, i) && isclr(prg->prog_used, i)) { + setbit(prg->prog_used, i); + f = array(prg->prog_fds, i, FILE*); + break; + } + pthread_mutex_unlock(&prg->prog_mtx); + + return f; +} + +/* + * io_progDetach() - Detch from open program + * + * @prg= program pool + * @pfd = attached program handle + * return: none + */ +void +io_progDetach(prog_t * __restrict prg, FILE *pfd) +{ + register int i; + + if (!prg || !pfd) + return; + + pthread_mutex_lock(&prg->prog_mtx); + for (i = 0; i < array_Size(prg->prog_fds); i++) + if (array(prg->prog_fds, i, FILE*) == pfd) { + clrbit(prg->prog_used, i); + break; + } + pthread_mutex_unlock(&prg->prog_mtx); }