1 /* Copyright (C) 2018-2023 Free Software Foundation, Inc.
2 Contributed by Nicolas Koenig
3
4 This file is part of the GNU Fortran runtime library (libgfortran).
5
6 Libgfortran is free software; you can redistribute it and/or modify
7 it under the terms of the GNU General Public License as published by
8 the Free Software Foundation; either version 3, or (at your option)
9 any later version.
10
11 Libgfortran is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 GNU General Public License for more details.
15
16 Under Section 7 of GPL version 3, you are granted additional
17 permissions described in the GCC Runtime Library Exception, version
18 3.1, as published by the Free Software Foundation.
19
20 You should have received a copy of the GNU General Public License and
21 a copy of the GCC Runtime Library Exception along with this program;
22 see the files COPYING3 and COPYING.RUNTIME respectively. If not, see
23 <http://www.gnu.org/licenses/>. */
24
25 #include "libgfortran.h"
26
27 #define _GTHREAD_USE_COND_INIT_FUNC
28 #include "../../libgcc/gthr.h"
29 #include "io.h"
30 #include "fbuf.h"
31 #include "format.h"
32 #include "unix.h"
33 #include <string.h>
34 #include <assert.h>
35
36 #include <sys/types.h>
37
38 #include "async.h"
39 #if ASYNC_IO
40
41 DEBUG_LINE (__thread const char *aio_prefix = MPREFIX);
42
43 DEBUG_LINE (__gthread_mutex_t debug_queue_lock = __GTHREAD_MUTEX_INIT;)
44 DEBUG_LINE (aio_lock_debug *aio_debug_head = NULL;)
45
46 /* Current unit for asynchronous I/O. Needed for error reporting. */
47
48 __thread gfc_unit *thread_unit = NULL;
49
50 /* Queue entry for the asynchronous I/O entry. */
51 typedef struct transfer_queue
52 {
53 enum aio_do type;
54 struct transfer_queue *next;
55 struct st_parameter_dt *new_pdt;
56 transfer_args arg;
57 _Bool has_id;
58 int read_flag;
59 } transfer_queue;
60
61 struct error {
62 st_parameter_dt *dtp;
63 int id;
64 };
65
66 /* Helper function to exchange the old vs. a new PDT. */
67
68 static void
69 update_pdt (st_parameter_dt **old, st_parameter_dt *new) {
70 st_parameter_dt *temp;
71 NOTE ("Changing pdts, current_unit = %p", (void *) (new->u.p.current_unit));
72 temp = *old;
73 *old = new;
74 if (temp)
75 free (temp);
76 }
77
78 /* Destroy an adv_cond structure. */
79
80 static void
81 destroy_adv_cond (struct adv_cond *ac)
82 {
83 T_ERROR (__gthread_cond_destroy, &ac->signal);
84 }
85
86 /* Function invoked as start routine for a new asynchronous I/O unit.
87 Contains the main loop for accepting requests and handling them. */
88
89 static void *
90 async_io (void *arg)
91 {
92 DEBUG_LINE (aio_prefix = TPREFIX);
93 transfer_queue *ctq = NULL, *prev = NULL;
94 gfc_unit *u = (gfc_unit *) arg;
95 async_unit *au = u->au;
96 LOCK (&au->lock);
97 thread_unit = u;
98 au->thread = __gthread_self ();
99 while (true)
100 {
101 /* Main loop. At this point, au->lock is always held. */
102 WAIT_SIGNAL_MUTEX (&au->work, au->tail != NULL, &au->lock);
103 LOCK (&au->lock);
104 ctq = au->head;
105 prev = NULL;
106 /* Loop over the queue entries until they are finished. */
107 while (ctq)
108 {
109 if (prev)
110 free (prev);
111 prev = ctq;
112 if (!au->error.has_error)
113 {
114 UNLOCK (&au->lock);
115
116 switch (ctq->type)
117 {
118 case AIO_WRITE_DONE:
119 NOTE ("Finalizing write");
120 st_write_done_worker (au->pdt, false);
121 UNLOCK (&au->io_lock);
122 break;
123
124 case AIO_READ_DONE:
125 NOTE ("Finalizing read");
126 st_read_done_worker (au->pdt, false);
127 UNLOCK (&au->io_lock);
128 break;
129
130 case AIO_DATA_TRANSFER_INIT:
131 NOTE ("Data transfer init");
132 LOCK (&au->io_lock);
133 update_pdt (&au->pdt, ctq->new_pdt);
134 data_transfer_init_worker (au->pdt, ctq->read_flag);
135 break;
136
137 case AIO_TRANSFER_SCALAR:
138 NOTE ("Starting scalar transfer");
139 ctq->arg.scalar.transfer (au->pdt, ctq->arg.scalar.arg_bt,
140 ctq->arg.scalar.data,
141 ctq->arg.scalar.i,
142 ctq->arg.scalar.s1,
143 ctq->arg.scalar.s2);
144 break;
145
146 case AIO_TRANSFER_ARRAY:
147 NOTE ("Starting array transfer");
148 NOTE ("ctq->arg.array.desc = %p",
149 (void *) (ctq->arg.array.desc));
150 transfer_array_inner (au->pdt, ctq->arg.array.desc,
151 ctq->arg.array.kind,
152 ctq->arg.array.charlen);
153 free (ctq->arg.array.desc);
154 break;
155
156 case AIO_CLOSE:
157 NOTE ("Received AIO_CLOSE");
158 LOCK (&au->lock);
159 goto finish_thread;
160
161 default:
162 internal_error (NULL, "Invalid queue type");
163 break;
164 }
165 LOCK (&au->lock);
166 if (unlikely (au->error.has_error))
167 au->error.last_good_id = au->id.low - 1;
168 }
169 else
170 {
171 if (ctq->type == AIO_WRITE_DONE || ctq->type == AIO_READ_DONE)
172 {
173 UNLOCK (&au->io_lock);
174 }
175 else if (ctq->type == AIO_CLOSE)
176 {
177 NOTE ("Received AIO_CLOSE during error condition");
178 goto finish_thread;
179 }
180 }
181
182 NOTE ("Next ctq, current id: %d", au->id.low);
183 if (ctq->has_id && au->id.waiting == au->id.low++)
184 SIGNAL (&au->id.done);
185
186 ctq = ctq->next;
187 }
188 au->tail = NULL;
189 au->head = NULL;
190 au->empty = 1;
191 SIGNAL (&au->emptysignal);
192 }
193 finish_thread:
194 au->tail = NULL;
195 au->head = NULL;
196 au->empty = 1;
197 SIGNAL (&au->emptysignal);
198 free (ctq);
199 UNLOCK (&au->lock);
200 return NULL;
201 }
202
203 /* Free an asynchronous unit. */
204
205 static void
206 free_async_unit (async_unit *au)
207 {
208 if (au->tail)
209 internal_error (NULL, "Trying to free nonempty asynchronous unit");
210
211 destroy_adv_cond (&au->work);
212 destroy_adv_cond (&au->emptysignal);
213 destroy_adv_cond (&au->id.done);
214 T_ERROR (__gthread_mutex_destroy, &au->lock);
215 free (au);
216 }
217
218 /* Initialize an adv_cond structure. */
219
220 static void
221 init_adv_cond (struct adv_cond *ac)
222 {
223 ac->pending = 0;
224 __GTHREAD_COND_INIT_FUNCTION (&ac->signal);
225 }
226
227 /* Initialize an asyncronous unit, returning zero on success,
228 nonzero on failure. It also sets u->au. */
229
230 void
231 init_async_unit (gfc_unit *u)
232 {
233 async_unit *au;
234 if (!__gthread_active_p ())
235 {
236 u->au = NULL;
237 return;
238 }
239
240 au = (async_unit *) xmalloc (sizeof (async_unit));
241 u->au = au;
242 init_adv_cond (&au->work);
243 init_adv_cond (&au->emptysignal);
244 __GTHREAD_MUTEX_INIT_FUNCTION (&au->lock);
245 __GTHREAD_MUTEX_INIT_FUNCTION (&au->io_lock);
246 LOCK (&au->lock);
247 T_ERROR (__gthread_create, &au->thread, &async_io, (void *) u);
248 au->pdt = NULL;
249 au->head = NULL;
250 au->tail = NULL;
251 au->empty = true;
252 au->id.waiting = -1;
253 au->id.low = 0;
254 au->id.high = 0;
255 au->error.fatal_error = 0;
256 au->error.has_error = 0;
257 au->error.last_good_id = 0;
258 init_adv_cond (&au->id.done);
259 UNLOCK (&au->lock);
260 }
261
262 /* Enqueue a transfer statement. */
263
264 void
265 enqueue_transfer (async_unit *au, transfer_args *arg, enum aio_do type)
266 {
267 transfer_queue *tq = calloc (sizeof (transfer_queue), 1);
268 tq->arg = *arg;
269 tq->type = type;
270 tq->has_id = 0;
271 LOCK (&au->lock);
272 if (!au->tail)
273 au->head = tq;
274 else
275 au->tail->next = tq;
276 au->tail = tq;
277 REVOKE_SIGNAL (&(au->emptysignal));
278 au->empty = false;
279 SIGNAL (&au->work);
280 UNLOCK (&au->lock);
281 }
282
283 /* Enqueue an st_write_done or st_read_done which contains an ID. */
284
285 int
286 enqueue_done_id (async_unit *au, enum aio_do type)
287 {
288 int ret;
289 transfer_queue *tq = calloc (sizeof (transfer_queue), 1);
290
291 tq->type = type;
292 tq->has_id = 1;
293 LOCK (&au->lock);
294 if (!au->tail)
295 au->head = tq;
296 else
297 au->tail->next = tq;
298 au->tail = tq;
299 REVOKE_SIGNAL (&(au->emptysignal));
300 au->empty = false;
301 ret = au->id.high++;
302 NOTE ("Enqueue id: %d", ret);
303 SIGNAL (&au->work);
304 UNLOCK (&au->lock);
305 return ret;
306 }
307
308 /* Enqueue an st_write_done or st_read_done without an ID. */
309
310 void
311 enqueue_done (async_unit *au, enum aio_do type)
312 {
313 transfer_queue *tq = calloc (sizeof (transfer_queue), 1);
314 tq->type = type;
315 tq->has_id = 0;
316 LOCK (&au->lock);
317 if (!au->tail)
318 au->head = tq;
319 else
320 au->tail->next = tq;
321 au->tail = tq;
322 REVOKE_SIGNAL (&(au->emptysignal));
323 au->empty = false;
324 SIGNAL (&au->work);
325 UNLOCK (&au->lock);
326 }
327
328 /* Enqueue a CLOSE statement. */
329
330 void
331 enqueue_close (async_unit *au)
332 {
333 transfer_queue *tq = calloc (sizeof (transfer_queue), 1);
334
335 tq->type = AIO_CLOSE;
336 LOCK (&au->lock);
337 if (!au->tail)
338 au->head = tq;
339 else
340 au->tail->next = tq;
341 au->tail = tq;
342 REVOKE_SIGNAL (&(au->emptysignal));
343 au->empty = false;
344 SIGNAL (&au->work);
345 UNLOCK (&au->lock);
346 }
347
348 /* The asynchronous unit keeps the currently active PDT around.
349 This function changes that to the current one. */
350
351 void
352 enqueue_data_transfer_init (async_unit *au, st_parameter_dt *dt, int read_flag)
353 {
354 st_parameter_dt *new = xmalloc (sizeof (st_parameter_dt));
355 transfer_queue *tq = xmalloc (sizeof (transfer_queue));
356
357 memcpy ((void *) new, (void *) dt, sizeof (st_parameter_dt));
358
359 NOTE ("dt->internal_unit_desc = %p", dt->internal_unit_desc);
360 NOTE ("common.flags & mask = %d", dt->common.flags & IOPARM_LIBRETURN_MASK);
361 tq->next = NULL;
362 tq->type = AIO_DATA_TRANSFER_INIT;
363 tq->read_flag = read_flag;
364 tq->has_id = 0;
365 tq->new_pdt = new;
366 LOCK (&au->lock);
367
368 if (!au->tail)
369 au->head = tq;
370 else
371 au->tail->next = tq;
372 au->tail = tq;
373 REVOKE_SIGNAL (&(au->emptysignal));
374 au->empty = false;
375 SIGNAL (&au->work);
376 UNLOCK (&au->lock);
377 }
378
379 /* Collect the errors that may have happened asynchronously. Return true if
380 an error has been encountered. */
381
382 bool
383 collect_async_errors (st_parameter_common *cmp, async_unit *au)
384 {
385 bool has_error = au->error.has_error;
386
387 if (has_error)
388 {
389 if (generate_error_common (cmp, au->error.family, au->error.message))
390 {
391 au->error.has_error = 0;
392 au->error.cmp = NULL;
393 }
394 else
395 {
396 /* The program will exit later. */
397 au->error.fatal_error = true;
398 }
399 }
400 return has_error;
401 }
402
403 /* Perform a wait operation on an asynchronous unit with an ID specified,
404 which means collecting the errors that may have happened asynchronously.
405 Return true if an error has been encountered. */
406
407 bool
408 async_wait_id (st_parameter_common *cmp, async_unit *au, int i)
409 {
410 bool ret;
411
412 if (au == NULL)
413 return false;
414
415 if (cmp == NULL)
416 cmp = au->error.cmp;
417
418 if (au->error.has_error)
419 {
420 if (i <= au->error.last_good_id)
421 return false;
422
423 return collect_async_errors (cmp, au);
424 }
425
426 LOCK (&au->lock);
427 if (i > au->id.high)
428 {
429 generate_error_common (cmp, LIBERROR_BAD_WAIT_ID, NULL);
430 UNLOCK (&au->lock);
431 return true;
432 }
433
434 NOTE ("Waiting for id %d", i);
435 if (au->id.waiting < i)
436 au->id.waiting = i;
437 SIGNAL (&(au->work));
438 WAIT_SIGNAL_MUTEX (&(au->id.done),
439 (au->id.low >= au->id.waiting || au->empty), &au->lock);
440 LOCK (&au->lock);
441 ret = collect_async_errors (cmp, au);
442 UNLOCK (&au->lock);
443 return ret;
444 }
445
446 /* Perform a wait operation an an asynchronous unit without an ID. */
447
448 bool
449 async_wait (st_parameter_common *cmp, async_unit *au)
450 {
451 bool ret;
452
453 if (au == NULL)
454 return false;
455
456 if (cmp == NULL)
457 cmp = au->error.cmp;
458
459 LOCK (&(au->lock));
460 SIGNAL (&(au->work));
461
462 if (au->empty)
463 {
464 ret = collect_async_errors (cmp, au);
465 UNLOCK (&au->lock);
466 return ret;
467 }
468
469 WAIT_SIGNAL_MUTEX (&(au->emptysignal), (au->empty), &au->lock);
470 ret = collect_async_errors (cmp, au);
471 return ret;
472 }
473
474 /* Close an asynchronous unit. */
475
476 void
477 async_close (async_unit *au)
478 {
479 if (au == NULL)
480 return;
481
482 NOTE ("Closing async unit");
483 enqueue_close (au);
484 T_ERROR (__gthread_join, au->thread, NULL);
485 free_async_unit (au);
486 }
487
488 #else
489
490 /* Only set u->au to NULL so no async I/O will happen. */
491
492 void
493 init_async_unit (gfc_unit *u)
494 {
495 u->au = NULL;
496 return;
497 }
498
499 /* Do-nothing function, which will not be called. */
500
501 void
502 enqueue_transfer (async_unit *au, transfer_args *arg, enum aio_do type)
503 {
504 return;
505 }
506
507 /* Do-nothing function, which will not be called. */
508
509 int
510 enqueue_done_id (async_unit *au, enum aio_do type)
511 {
512 return 0;
513 }
514
515 /* Do-nothing function, which will not be called. */
516
517 void
518 enqueue_done (async_unit *au, enum aio_do type)
519 {
520 return;
521 }
522
523 /* Do-nothing function, which will not be called. */
524
525 void
526 enqueue_close (async_unit *au)
527 {
528 return;
529 }
530
531 /* Do-nothing function, which will not be called. */
532
533 void
534 enqueue_data_transfer_init (async_unit *au, st_parameter_dt *dt, int read_flag)
535 {
536 return;
537 }
538
539 /* Do-nothing function, which will not be called. */
540
541 bool
542 collect_async_errors (st_parameter_common *cmp, async_unit *au)
543 {
544 return false;
545 }
546
547 /* Do-nothing function, which will not be called. */
548
549 bool
550 async_wait_id (st_parameter_common *cmp, async_unit *au, int i)
551 {
552 return false;
553 }
554
555 /* Do-nothing function, which will not be called. */
556
557 bool
558 async_wait (st_parameter_common *cmp, async_unit *au)
559 {
560 return false;
561 }
562
563 /* Do-nothing function, which will not be called. */
564
565 void
566 async_close (async_unit *au)
567 {
568 return;
569 }
570
571 #endif