server: Implement io completion messages for job objects.

dlls/kernel32/tests/process.c

@@ -2387,7 +2387,6 @@ static void test_CompletionPort(void)
     port_info.CompletionKey = job;
     port_info.CompletionPort = port;
     ret = pSetInformationJobObject(job, JobObjectAssociateCompletionPortInformation, &port_info, sizeof(port_info));
-    todo_wine
     ok(ret, "SetInformationJobObject error %u\n", GetLastError());
 
     create_process("wait", &pi);
@@ -2395,16 +2394,13 @@ static void test_CompletionPort(void)
     ret = pAssignProcessToJobObject(job, pi.hProcess);
     ok(ret, "AssignProcessToJobObject error %u\n", GetLastError());
 
-    todo_wine
     test_completion(port, JOB_OBJECT_MSG_NEW_PROCESS, (DWORD_PTR)job, pi.dwProcessId, 0);
 
     TerminateProcess(pi.hProcess, 0);
     dwret = WaitForSingleObject(pi.hProcess, 1000);
     ok(dwret == WAIT_OBJECT_0, "WaitForSingleObject returned %u\n", dwret);
 
-    todo_wine
     test_completion(port, JOB_OBJECT_MSG_EXIT_PROCESS, (DWORD_PTR)job, pi.dwProcessId, 0);
-    todo_wine
     test_completion(port, JOB_OBJECT_MSG_ACTIVE_PROCESS_ZERO, (DWORD_PTR)job, 0, 100);
 
     CloseHandle(pi.hProcess);

dlls/ntdll/sync.c

@@ -668,6 +668,21 @@ NTSTATUS WINAPI NtSetInformationJobObject( HANDLE handle, JOBOBJECTINFOCLASS cla
         SERVER_END_REQ;
         break;
 
+    case JobObjectAssociateCompletionPortInformation:
+        if (len != sizeof(JOBOBJECT_ASSOCIATE_COMPLETION_PORT))
+            return STATUS_INVALID_PARAMETER;
+
+        SERVER_START_REQ( set_job_completion_port )
+        {
+            JOBOBJECT_ASSOCIATE_COMPLETION_PORT *port_info = info;
+            req->job = wine_server_obj_handle( handle );
+            req->port = wine_server_obj_handle( port_info->CompletionPort );
+            req->key = wine_server_client_ptr( port_info->CompletionKey );
+            status = wine_server_call(req);
+        }
+        SERVER_END_REQ;
+        break;
+
     default:
         FIXME( "stub: %p %u %p %u\n", handle, class, info, len );
         return STATUS_NOT_IMPLEMENTED;

include/wine/server_protocol.h

@@ -5141,6 +5141,21 @@ struct set_job_limits_reply
 };
 
 
+
+struct set_job_completion_port_request
+{
+    struct request_header __header;
+    obj_handle_t job;
+    obj_handle_t port;
+    char __pad_20[4];
+    client_ptr_t key;
+};
+struct set_job_completion_port_reply
+{
+    struct reply_header __header;
+};
+
+
 enum request
 {
     REQ_new_process,
@@ -5403,6 +5418,7 @@ enum request
     REQ_assign_job,
     REQ_process_in_job,
     REQ_set_job_limits,
+    REQ_set_job_completion_port,
     REQ_NB_REQUESTS
 };
 
@@ -5670,6 +5686,7 @@ union generic_request
     struct assign_job_request assign_job_request;
     struct process_in_job_request process_in_job_request;
     struct set_job_limits_request set_job_limits_request;
+    struct set_job_completion_port_request set_job_completion_port_request;
 };
 union generic_reply
 {
@@ -5935,8 +5952,9 @@ union generic_reply
     struct assign_job_reply assign_job_reply;
     struct process_in_job_reply process_in_job_reply;
     struct set_job_limits_reply set_job_limits_reply;
+    struct set_job_completion_port_reply set_job_completion_port_reply;
 };
 
-#define SERVER_PROTOCOL_VERSION 464
+#define SERVER_PROTOCOL_VERSION 465
 
 #endif /* __WINE_WINE_SERVER_PROTOCOL_H */

server/process.c

@@ -147,6 +147,8 @@ struct job
     struct list process_list;      /* list of all processes */
     int num_processes;             /* count of running processes */
     unsigned int limit_flags;      /* limit flags */
+    struct completion *completion_port; /* associated completion port */
+    apc_param_t completion_key;    /* key to send with completion messages */
 };
 
 static const struct object_ops job_ops =
@@ -186,6 +188,8 @@ static struct job *create_job_object( struct directory *root, const struct unico
             list_init( &job->process_list );
             job->num_processes = 0;
             job->limit_flags = 0;
+            job->completion_port = NULL;
+            job->completion_key = 0;
         }
     }
     return job;
@@ -212,6 +216,12 @@ static unsigned int job_map_access( struct object *obj, unsigned int access )
     return access & ~(GENERIC_READ | GENERIC_WRITE | GENERIC_EXECUTE | GENERIC_ALL);
 }
 
+static void add_job_completion( struct job *job, apc_param_t msg, apc_param_t pid )
+{
+    if (job->completion_port)
+        add_completion( job->completion_port, job->completion_key, pid, STATUS_SUCCESS, msg );
+}
+
 static void add_job_process( struct job *job, struct process *process )
 {
     if (!process->running_threads)
@@ -227,6 +237,8 @@ static void add_job_process( struct job *job, struct process *process )
     process->job = (struct job *)grab_object( job );
     list_add_tail( &job->process_list, &process->job_entry );
     job->num_processes++;
+
+    add_job_completion( job, JOB_OBJECT_MSG_NEW_PROCESS, get_process_id(process) );
 }
 
 /* called when a process has terminated, allow one additional process */
@@ -238,6 +250,11 @@ static void release_job_process( struct process *process )
 
     assert( job->num_processes );
     job->num_processes--;
+
+    add_job_completion( job, JOB_OBJECT_MSG_EXIT_PROCESS, get_process_id(process) );
+
+    if (!job->num_processes)
+        add_job_completion( job, JOB_OBJECT_MSG_ACTIVE_PROCESS_ZERO, 0 );
 }
 
 static void job_destroy( struct object *obj )
@@ -247,6 +264,8 @@ static void job_destroy( struct object *obj )
 
     assert( !job->num_processes );
     assert( list_empty(&job->process_list) );
+
+    if (job->completion_port) release_object( job->completion_port );
 }
 
 static void job_dump( struct object *obj, int verbose )
@@ -1534,3 +1553,21 @@ DECL_HANDLER(set_job_limits)
     job->limit_flags = req->limit_flags;
     release_object( job );
 }
+
+/* set the jobs completion port */
+DECL_HANDLER(set_job_completion_port)
+{
+    struct job *job = get_job_obj( current->process, req->job, JOB_OBJECT_SET_ATTRIBUTES );
+
+    if (!job) return;
+
+    if (!job->completion_port)
+    {
+        job->completion_port = get_completion_obj( current->process, req->port, IO_COMPLETION_MODIFY_STATE );
+        job->completion_key = req->key;
+    }
+    else
+        set_error( STATUS_INVALID_PARAMETER );
+
+    release_object( job );
+}

server/protocol.def

@@ -3535,3 +3535,11 @@ enum coords_relative
     obj_handle_t handle;          /* handle to the job */
     unsigned int limit_flags;     /* new limit flags */
 @END
+
+
+/* Set new completion port for a job */
+@REQ(set_job_completion_port)
+    obj_handle_t job;             /* handle to the job */
+    obj_handle_t port;            /* handle to the completion port */
+    client_ptr_t key;             /* key to send with completion messages */
+@END

server/request.h

@@ -366,6 +366,7 @@ DECL_HANDLER(create_job);
 DECL_HANDLER(assign_job);
 DECL_HANDLER(process_in_job);
 DECL_HANDLER(set_job_limits);
+DECL_HANDLER(set_job_completion_port);
 
 #ifdef WANT_REQUEST_HANDLERS
 
@@ -632,6 +633,7 @@ static const req_handler req_handlers[REQ_NB_REQUESTS] =
     (req_handler)req_assign_job,
     (req_handler)req_process_in_job,
     (req_handler)req_set_job_limits,
+    (req_handler)req_set_job_completion_port,
 };
 
 C_ASSERT( sizeof(affinity_t) == 8 );
@@ -2226,6 +2228,10 @@ C_ASSERT( sizeof(struct process_in_job_request) == 24 );
 C_ASSERT( FIELD_OFFSET(struct set_job_limits_request, handle) == 12 );
 C_ASSERT( FIELD_OFFSET(struct set_job_limits_request, limit_flags) == 16 );
 C_ASSERT( sizeof(struct set_job_limits_request) == 24 );
+C_ASSERT( FIELD_OFFSET(struct set_job_completion_port_request, job) == 12 );
+C_ASSERT( FIELD_OFFSET(struct set_job_completion_port_request, port) == 16 );
+C_ASSERT( FIELD_OFFSET(struct set_job_completion_port_request, key) == 24 );
+C_ASSERT( sizeof(struct set_job_completion_port_request) == 32 );
 
 #endif  /* WANT_REQUEST_HANDLERS */
 

server/trace.c

@@ -4125,6 +4125,13 @@ static void dump_set_job_limits_request( const struct set_job_limits_request *re
     fprintf( stderr, ", limit_flags=%08x", req->limit_flags );
 }
 
+static void dump_set_job_completion_port_request( const struct set_job_completion_port_request *req )
+{
+    fprintf( stderr, " job=%04x", req->job );
+    fprintf( stderr, ", port=%04x", req->port );
+    dump_uint64( ", key=", &req->key );
+}
+
 static const dump_func req_dumpers[REQ_NB_REQUESTS] = {
     (dump_func)dump_new_process_request,
     (dump_func)dump_get_new_process_info_request,
@@ -4386,6 +4393,7 @@ static const dump_func req_dumpers[REQ_NB_REQUESTS] = {
     (dump_func)dump_assign_job_request,
     (dump_func)dump_process_in_job_request,
     (dump_func)dump_set_job_limits_request,
+    (dump_func)dump_set_job_completion_port_request,
 };
 
 static const dump_func reply_dumpers[REQ_NB_REQUESTS] = {
@@ -4649,6 +4657,7 @@ static const dump_func reply_dumpers[REQ_NB_REQUESTS] = {
     NULL,
     NULL,
     NULL,
+    NULL,
 };
 
 static const char * const req_names[REQ_NB_REQUESTS] = {
@@ -4912,6 +4921,7 @@ static const char * const req_names[REQ_NB_REQUESTS] = {
     "assign_job",
     "process_in_job",
     "set_job_limits",
+    "set_job_completion_port",
 };
 
 static const struct