Line data Source code
1 : /*
2 : *
3 : * Copyright 2015, Google Inc.
4 : * All rights reserved.
5 : *
6 : * Redistribution and use in source and binary forms, with or without
7 : * modification, are permitted provided that the following conditions are
8 : * met:
9 : *
10 : * * Redistributions of source code must retain the above copyright
11 : * notice, this list of conditions and the following disclaimer.
12 : * * Redistributions in binary form must reproduce the above
13 : * copyright notice, this list of conditions and the following disclaimer
14 : * in the documentation and/or other materials provided with the
15 : * distribution.
16 : * * Neither the name of Google Inc. nor the names of its
17 : * contributors may be used to endorse or promote products derived from
18 : * this software without specific prior written permission.
19 : *
20 : * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 : * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 : * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 : * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 : * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 : * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 : * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 : * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 : * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 : * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 : * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31 : *
32 : */
33 :
34 : #include <grpc/support/port_platform.h>
35 :
36 : #ifdef GPR_POSIX_SOCKET
37 :
38 : #include "src/core/iomgr/workqueue.h"
39 :
40 : #include <stdio.h>
41 :
42 : #include <grpc/support/alloc.h>
43 : #include <grpc/support/log.h>
44 : #include <grpc/support/useful.h>
45 :
46 : #include "src/core/iomgr/fd_posix.h"
47 :
48 : static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, int success);
49 :
50 1 : grpc_workqueue *grpc_workqueue_create(grpc_exec_ctx *exec_ctx) {
51 : char name[32];
52 1 : grpc_workqueue *workqueue = gpr_malloc(sizeof(grpc_workqueue));
53 1 : gpr_ref_init(&workqueue->refs, 1);
54 1 : gpr_mu_init(&workqueue->mu);
55 1 : workqueue->closure_list.head = workqueue->closure_list.tail = NULL;
56 1 : grpc_wakeup_fd_init(&workqueue->wakeup_fd);
57 1 : sprintf(name, "workqueue:%p", (void *)workqueue);
58 1 : workqueue->wakeup_read_fd =
59 1 : grpc_fd_create(GRPC_WAKEUP_FD_GET_READ_FD(&workqueue->wakeup_fd), name);
60 1 : grpc_closure_init(&workqueue->read_closure, on_readable, workqueue);
61 1 : grpc_fd_notify_on_read(exec_ctx, workqueue->wakeup_read_fd,
62 : &workqueue->read_closure);
63 1 : return workqueue;
64 : }
65 :
66 1 : static void workqueue_destroy(grpc_exec_ctx *exec_ctx,
67 : grpc_workqueue *workqueue) {
68 1 : GPR_ASSERT(grpc_closure_list_empty(workqueue->closure_list));
69 1 : grpc_fd_shutdown(exec_ctx, workqueue->wakeup_read_fd);
70 1 : }
71 :
72 : #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG
73 0 : void grpc_workqueue_ref(grpc_workqueue *workqueue, const char *file, int line,
74 : const char *reason) {
75 0 : gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "WORKQUEUE:%p ref %d -> %d %s",
76 0 : workqueue, (int)workqueue->refs.count, (int)workqueue->refs.count + 1,
77 : reason);
78 : #else
79 : void grpc_workqueue_ref(grpc_workqueue *workqueue) {
80 : #endif
81 0 : gpr_ref(&workqueue->refs);
82 0 : }
83 :
84 : #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG
85 1 : void grpc_workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue,
86 : const char *file, int line, const char *reason) {
87 2 : gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "WORKQUEUE:%p unref %d -> %d %s",
88 2 : workqueue, (int)workqueue->refs.count, (int)workqueue->refs.count - 1,
89 : reason);
90 : #else
91 : void grpc_workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) {
92 : #endif
93 1 : if (gpr_unref(&workqueue->refs)) {
94 1 : workqueue_destroy(exec_ctx, workqueue);
95 : }
96 1 : }
97 :
98 1 : void grpc_workqueue_add_to_pollset(grpc_exec_ctx *exec_ctx,
99 : grpc_workqueue *workqueue,
100 : grpc_pollset *pollset) {
101 1 : grpc_pollset_add_fd(exec_ctx, pollset, workqueue->wakeup_read_fd);
102 1 : }
103 :
104 0 : void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) {
105 0 : gpr_mu_lock(&workqueue->mu);
106 0 : grpc_closure_list_move(&exec_ctx->closure_list, &workqueue->closure_list);
107 0 : gpr_mu_unlock(&workqueue->mu);
108 0 : }
109 :
110 2 : static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, int success) {
111 2 : grpc_workqueue *workqueue = arg;
112 :
113 2 : if (!success) {
114 1 : gpr_mu_destroy(&workqueue->mu);
115 : /* HACK: let wakeup_fd code know that we stole the fd */
116 1 : workqueue->wakeup_fd.read_fd = 0;
117 1 : grpc_wakeup_fd_destroy(&workqueue->wakeup_fd);
118 1 : grpc_fd_orphan(exec_ctx, workqueue->wakeup_read_fd, NULL, "destroy");
119 1 : gpr_free(workqueue);
120 : } else {
121 1 : gpr_mu_lock(&workqueue->mu);
122 1 : grpc_closure_list_move(&workqueue->closure_list, &exec_ctx->closure_list);
123 1 : grpc_wakeup_fd_consume_wakeup(&workqueue->wakeup_fd);
124 1 : gpr_mu_unlock(&workqueue->mu);
125 1 : grpc_fd_notify_on_read(exec_ctx, workqueue->wakeup_read_fd,
126 : &workqueue->read_closure);
127 : }
128 2 : }
129 :
130 1 : void grpc_workqueue_push(grpc_workqueue *workqueue, grpc_closure *closure,
131 : int success) {
132 1 : closure->success = success;
133 1 : closure->next = NULL;
134 1 : gpr_mu_lock(&workqueue->mu);
135 1 : if (grpc_closure_list_empty(workqueue->closure_list)) {
136 1 : grpc_wakeup_fd_wakeup(&workqueue->wakeup_fd);
137 : }
138 1 : grpc_closure_list_add(&workqueue->closure_list, closure, success);
139 1 : gpr_mu_unlock(&workqueue->mu);
140 1 : }
141 :
142 : #endif /* GPR_POSIX_SOCKET */
|
