The R Project SVN – R – Diff – /trunk/src/gnuwin32/run.c

 /*
  *  R : A Computer Language for Statistical Data Analysis
  *  file run.c: a simple 'reading' pipe (and a command executor)
  *  Copyright  (C) 1999-2001  Guido Masarotto and Brian Ripley
- *             (C) 2007-2017  The R Core Team
+ *             (C) 2007-2019  The R Core Team
+ *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
  *  the Free Software Foundation; either version 2 of the License, or
  *  (at your option) any later version.
 extern size_t Rf_utf8towcs(wchar_t *wc, const char *s, size_t n);
 static void pcreate(const char* cmd, cetype_t enc,
 		      int newconsole, int visible,
 		      HANDLE hIN, HANDLE hOUT, HANDLE hERR,
-		      PROCESS_INFORMATION *pi)
+		      pinfo *pi)
+{
     DWORD ret;
     STARTUPINFO si;
     STARTUPINFOW wsi;
-    HANDLE dupIN, dupOUT, dupERR;
+    HANDLE dupIN, dupOUT, dupERR, job, port;
     WORD showWindow = SW_SHOWDEFAULT;
+    DWORD flags;
+    BOOL inJob;
+    Rboolean breakaway;
+    JOBOBJECT_EXTENDED_LIMIT_INFORMATION jeli;
+    JOBOBJECT_ASSOCIATE_COMPLETION_PORT cport;
     int inpipe;
     char *ecmd;
     SECURITY_ATTRIBUTES sa;
     sa.nLength = sizeof(sa);
     sa.lpSecurityDescriptor = NULL;
 	    si.hStdOutput = dupOUT;
 	    si.hStdError  = dupERR;
+	}
+    }
+    /* Originally, the external process has been waited for only using
+       waitForSingleObject, but that has been proven unreliable: sometimes
+       the output file would still be opened (and hence locked) by some
+       child process after waitForSingleObject would finish. This has been
+       observed also while running tests and particularly when building
+       vignettes, resulting in spurious "Permission denied" errors.
+       This has been happening almost surely due to a child process not
+       waiting for its own children to finish, which has been reported
+       to happen with Linux utilities ported to Windows as used for tests
+       in Haskell/GHC. Inspired by Haskell process and a blog post about
+       waiting for a process tree to finish, we now use job objects to
+       wait also for process trees with this issue:
+	https://github.com/haskell/process
+	https://blogs.msdn.microsoft.com/oldnewthing/20130405-00/?p=4743
+       In addition, we try to be easy on applications coded to rely on that
+       they do not run in a job, when running in old Windows that do not
+       support nested jobs. With nested jobs support, it might make sense
+       to not breakaway to better support nested R processes.
+    */
+    /* Creating the process with CREATE_BREAKAWAY_FROM_JOB is safe when
+       the process is not in any job or when it is in a job that allows it.
+       The documentation does not say what would happen if we set the flag,
+       but run in a job that does not allow it, so better don't. */
+    breakaway = FALSE;
+    if (IsProcessInJob(GetCurrentProcess(), NULL, &inJob) && inJob) {
+	/* The documentation does not say that it would be ok to use
+	   QueryInformationJobObject when the process is not in the job,
+	   so we have better tested that upfront. */
+	ZeroMemory(&jeli, sizeof(JOBOBJECT_EXTENDED_LIMIT_INFORMATION));
+	ret = QueryInformationJobObject(
+		NULL,
+	        JobObjectExtendedLimitInformation,
+	        &jeli,
+	        sizeof(JOBOBJECT_EXTENDED_LIMIT_INFORMATION),
+		NULL);
+	breakaway = ret &&
+		(jeli.BasicLimitInformation.LimitFlags &
+	         JOB_OBJECT_LIMIT_BREAKAWAY_OK);
+    }
+    /* create a job that allows breakaway */
+    job = CreateJobObject(NULL, NULL);
+    if (job) {
+	ZeroMemory(&jeli, sizeof(JOBOBJECT_EXTENDED_LIMIT_INFORMATION));
+	jeli.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_BREAKAWAY_OK;
+	ret = SetInformationJobObject(
+		job,
+		JobObjectExtendedLimitInformation,
+		&jeli,
+                sizeof(JOBOBJECT_EXTENDED_LIMIT_INFORMATION));
+	if (!ret) {
+	    CloseHandle(job);
+	    job = NULL;
+	}
+    }
+    /* create a completion port to learn when processes exit */
+    if (job) {
+	port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 1);
+	if (!port) {
+	    CloseHandle(job);
+	    job = NULL;
+	}
+    }
+    if (job) {
+	ZeroMemory(&cport, sizeof(JOBOBJECT_ASSOCIATE_COMPLETION_PORT));
+	cport.CompletionKey = job; /* use job handle as key */
+	cport.CompletionPort = port;
+	ret = SetInformationJobObject(
+	    job,
+	    JobObjectAssociateCompletionPortInformation,
+	    &cport,
+	    sizeof(JOBOBJECT_ASSOCIATE_COMPLETION_PORT));
+	if (!ret) {
+	    CloseHandle(job);
+	    CloseHandle(port);
+	    job = NULL;
+	}
+    }
+    flags = 0;
+    if (job)
+	flags |= CREATE_SUSPENDED; /* assign to job before it runs */
+    if (newconsole && (visible == 1))
+	flags |= CREATE_NEW_CONSOLE;
+    if (job && breakaway)
+	flags |= CREATE_BREAKAWAY_FROM_JOB;
     if(enc == CE_UTF8) {
 	int n = strlen(ecmd); /* max no of chars */
 	wchar_t wcmd[n+1];
 	Rf_utf8towcs(wcmd, ecmd, n+1);
-	ret = CreateProcessW(NULL, wcmd, &sa, &sa, TRUE,
+	ret = CreateProcessW(NULL, wcmd, &sa, &sa, TRUE, flags,
-			     (newconsole && (visible == 1)) ?
-			     CREATE_NEW_CONSOLE : 0,
-			     NULL, NULL, &wsi, pi);
+			     NULL, NULL, &wsi, &(pi->pi));
     } else
-	ret = CreateProcess(NULL, ecmd, &sa, &sa, TRUE,
+	ret = CreateProcess(NULL, ecmd, &sa, &sa, TRUE, flags,
-			    (newconsole && (visible == 1)) ?
+			    NULL, NULL, &si, &(pi->pi));
+    if (ret && job) {
+	/* process was created as suspended */
+	if (!AssignProcessToJobObject(job, pi->pi.hProcess)) {
+	    /* will fail running on Windows without support for nested jobs,
+	       when running in a job that does not allow breakaway */
+	    CloseHandle(job);
+	    CloseHandle(port);
+	    job = NULL;
+	}
+	ResumeThread(pi->pi.hThread);
+    }
+    if (ret && job) {
+	/* process is running in new job */
+	pi->job = job;
+	pi->port = port;
+    } else {
+	if (job) {
+	    CloseHandle(job);
-			    CREATE_NEW_CONSOLE : 0,
+	    CloseHandle(port);
-			    NULL, NULL, &si, pi);
+	    job = NULL;
+	}
+	pi->job = NULL;
+	pi->port = NULL;
+    }
     if (inpipe) {
 	CloseHandle(dupIN);
 	CloseHandle(dupOUT);
 	CloseHandle(dupERR);
+    }
     if (!ret)
 	snprintf(RunError, 500, _("'CreateProcess' failed to run '%s'"), ecmd);
-    else CloseHandle(pi->hThread);
+    else CloseHandle(pi->pi.hThread);
     free(ecmd);
     return;
+}
 /* used in rpipeOpen */
 threadedwait(LPVOID param)
+{
     rpipe *p = (rpipe *) param;
     if (p->timeoutMillis) {
-	DWORD wres = WaitForSingleObject(p->pi.hProcess, p->timeoutMillis);
+	DWORD wres = WaitForSingleObject(p->pi.pi.hProcess, p->timeoutMillis);
 	if (wres == WAIT_TIMEOUT) {
-	    TerminateProcess(p->pi.hProcess, 124);
+	    TerminateProcess(p->pi.pi.hProcess, 124);
 	    p->timedout = 1;
 	    /* wait up to 10s for the  process to actually terminate */
-	    WaitForSingleObject(p->pi.hProcess, 10000);
+	    WaitForSingleObject(p->pi.pi.hProcess, 10000);
+	}
     } else
-	WaitForSingleObject(p->pi.hProcess, INFINITE);
+	WaitForSingleObject(p->pi.pi.hProcess, INFINITE);
     DWORD ret;
-    GetExitCodeProcess(p->pi.hProcess, &ret);
+    GetExitCodeProcess(p->pi.pi.hProcess, &ret);
     p->exitcode = ret;
     FlushFileBuffers(p->write);
     FlushFileBuffers(p->read);
     p->active = 0;
 /* Terminate the process pwait2 is waiting for. */
 extern void GA_askok(const char *info);
+static void waitForJob(pinfo *pi, DWORD timeoutMillis, int* timedout)
+{
+    DWORD code, ret;
+    ULONG_PTR key;
+    DWORD beforeMillis;
+    JOBOBJECT_BASIC_ACCOUNTING_INFORMATION jbai;
+    LPOVERLAPPED overlapped; /* not used */
+    DWORD queryMillis;
+    if (timeoutMillis)
+	beforeMillis = timeGetTime();
+    queryMillis = 0;
+    for(;;) {
+	ret = GetQueuedCompletionStatus(pi->port, &code, &key,
+					&overlapped, queryMillis);
+	if (ret && code == JOB_OBJECT_MSG_ACTIVE_PROCESS_ZERO &&
+	    (HANDLE)key == pi->job)
+	    break;
+	/* start with short query timeouts because notifications often get lost,
+	   this is essentially polling */
+	if (queryMillis == 0)
+	    queryMillis = 1;
+	else if (queryMillis < 100)
+	    queryMillis *= 2;
+	if (timeoutMillis && (timeGetTime() - beforeMillis >= timeoutMillis)) {
+	    if (timedout)
+		*timedout = 1;
+	    break;
+	}
+	/* Check also explicitly because notifications are documented to get
+	   lost and they often do. */
+	ZeroMemory(&jbai, sizeof(JOBOBJECT_BASIC_ACCOUNTING_INFORMATION));
+	ret = QueryInformationJobObject(
+		pi->job,
+		JobObjectBasicAccountingInformation,
+		&jbai,
+		sizeof(JOBOBJECT_BASIC_ACCOUNTING_INFORMATION),
+		NULL);
+	if (ret && jbai.ActiveProcesses == 0)
+	    break;
+    }
+    CloseHandle(pi->port);
+    CloseHandle(pi->job);
+}
 static void terminate_process(void *p)
+{
-    PROCESS_INFORMATION *pi = (PROCESS_INFORMATION*)p;
+    pinfo *pi = (pinfo*) p;
-    EnumWindows((WNDENUMPROC)TerminateWindow, (LPARAM)pi->dwProcessId);
+    EnumWindows((WNDENUMPROC)TerminateWindow, (LPARAM)pi->pi.dwProcessId);
-    if (WaitForSingleObject(pi->hProcess, 5000) == WAIT_TIMEOUT) {
+    if (WaitForSingleObject(pi->pi.hProcess, 5000) == WAIT_TIMEOUT) {
 	if (R_Interactive)
 	    GA_askok(_("Child process not responding.  R will terminate it."));
-	TerminateProcess(pi->hProcess, 99);
+	TerminateProcess(pi->pi.hProcess, 99);
+    }
+    if (pi->job)
+	waitForJob(pi, 2000, NULL);
+}
-static int pwait2(HANDLE p, DWORD timeoutMillis, int* timedout)
+static int pwait2(pinfo *pi, DWORD timeoutMillis, int* timedout)
+{
     DWORD ret;
     if (!timeoutMillis) {
-	while( WaitForSingleObject(p, 100) == WAIT_TIMEOUT )
+	while( WaitForSingleObject(pi->pi.hProcess, 100) == WAIT_TIMEOUT )
 	    R_CheckUserInterrupt();
     } else {
 	DWORD beforeMillis = timeGetTime();
-	while( WaitForSingleObject(p, 100) == WAIT_TIMEOUT ) {
+	while( WaitForSingleObject(pi->pi.hProcess, 100) == WAIT_TIMEOUT ) {
 	    R_CheckUserInterrupt();
 	    DWORD afterMillis = timeGetTime();
 	    if (afterMillis - beforeMillis >= timeoutMillis) {
-		TerminateProcess(p, 124);
+		TerminateProcess(pi->pi.hProcess, 124);
 		if (timedout)
 		    *timedout = 1;
 		/* wait up to 10s for the process to actually terminate */
-		WaitForSingleObject(p, 10000);
+		WaitForSingleObject(pi->pi.hProcess, 10000);
 		break;
+	    }
+	}
+    }
-    GetExitCodeProcess(p, &ret);
+    GetExitCodeProcess(pi->pi.hProcess, &ret);
+    if (pi->job)
+	waitForJob(pi, timeoutMillis, timedout);
     return ret;
+}
 /*
   Used for external commands in file.show() and edit(), and for
     if (!wait && timeout)
 	error("Timeout with background running processes is not supported.");
     HANDLE hIN = getInputHandle(fin), hOUT, hERR;
     int ret = 0;
-    PROCESS_INFORMATION pi;
+    pinfo pi;
     int close1 = 0, close2 = 0, close3 = 0;
     if (hIN && fin && fin[0]) close1 = 1;
     hOUT = getOutputHandle(fout, 0);
 	if (!hERR) return 1;
 	if (ferr && ferr[0]) close3 = 1;
+    }
-    memset(&pi, 0, sizeof(pi));
+    memset(&(pi.pi), 0, sizeof(PROCESS_INFORMATION));
     pcreate(cmd, enc, !wait, visible, hIN, hOUT, hERR, &pi);
-    if (pi.hProcess) {
+    if (pi.pi.hProcess) {
 	if (wait) {
 	    RCNTXT cntxt;
 	    begincontext(&cntxt, CTXT_CCODE, R_NilValue, R_BaseEnv, R_BaseEnv,
 		     R_NilValue, R_NilValue);
 	    cntxt.cend = &terminate_process;
 	    cntxt.cenddata = &pi;
 	    DWORD timeoutMillis = (DWORD) (1000*timeout);
-	    ret = pwait2(pi.hProcess, timeoutMillis, timedout);
+	    ret = pwait2(&pi, timeoutMillis, timedout);
 	    endcontext(&cntxt);
 	    snprintf(RunError, 501, _("Exit code was %d"), ret);
 	    ret &= 0xffff;
 	} else ret = 0;
-	CloseHandle(pi.hProcess);
+	CloseHandle(pi.pi.hProcess);
     } else {
     	ret = NOLAUNCH;
+    }
     if (close1) CloseHandle(hIN);
     if (close2) CloseHandle(hOUT);
     if (!(r = (rpipe *) malloc(sizeof(struct structRPIPE)))) {
 	strcpy(RunError, _("Insufficient memory (rpipeOpen)"));
 	return NULL;
+    }
     r->active = 0;
-    r->pi.hProcess = NULL;
+    r->pi.pi.hProcess = NULL;
+    r->pi.job = NULL;
     r->thread = NULL;
     r->timedout = 0;
     r->timeoutMillis = (DWORD) (1000*timeout);
     res = CreatePipe(&hReadPipe, &hWritePipe, NULL, 0);
     if (res == FALSE) {
 	/* This sends stdout and stderr to NUL: */
 	pcreate(cmd, enc, 1, visible,
 		r->read, INVALID_HANDLE_VALUE, INVALID_HANDLE_VALUE,
 		&(r->pi));
 	r->active = 1;
-	if (!r->pi.hProcess) return NULL; else return r;
+	if (!r->pi.pi.hProcess) return NULL; else return r;
+    }
     /* pipe for R to read from */
     hTHIS = GetCurrentProcess();
     r->write = hWritePipe;
     if (close1) CloseHandle(hIN);
     if (close2) CloseHandle(hOUT);
     if (close3) CloseHandle(hERR);
     r->active = 1;
-    if (!r->pi.hProcess)
+    if (!r->pi.pi.hProcess)
 	return NULL;
     if (!(r->thread = CreateThread(NULL, 0, threadedwait, r, 0, &id))) {
 	rpipeClose(r, NULL);
 	strcpy(RunError, "CreateThread failed");
 	return NULL;
     rpipeTerminate(r);
     /* threadedwait may have obtained the exit code of the pipe process,
        but also may have been terminated too early; retrieve the exit
        code again to avoid race condition */
     DWORD ret;
-    GetExitCodeProcess(r->pi.hProcess, &ret);
+    GetExitCodeProcess(r->pi.pi.hProcess, &ret);
     r->exitcode = ret;
     CloseHandle(r->read);
     CloseHandle(r->write);
-    CloseHandle(r->pi.hProcess);
+    CloseHandle(r->pi.pi.hProcess);
     i = r->exitcode;
     if (timedout)
 	*timedout = r->timedout;
     free(r);
     return i &= 0xffff;
+{
     rpipe *rp = ((RWpipeconn)con->private) ->rp;
     DWORD towrite = nitems * size, write, ret;
     if(!rp->active) return 0;
-    GetExitCodeProcess(rp->pi.hProcess, &ret);
+    GetExitCodeProcess(rp->pi.pi.hProcess, &ret);
     if(ret != STILL_ACTIVE) {
 	rp->active = 0;
 	warning("broken Windows pipe");
 	return 0;
+    }

The R Project SVN R

(root)/trunk/src/gnuwin32/run.c – Rev 73675 → 75992