1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
/* SPDX-License-Identifier: GPL-2.0 */
/** \file stdin.c Functions that deal with reading from stdin. */
#include "para.h"
#include "list.h"
#include "sched.h"
#include "fd.h"
#include "error.h"
#include "stdin.h"
#include "buffer_tree.h"
#include "string.h"
struct stdin_task {
/* Stdin is always the root of the buffer tree. */
struct btr_node *btrn;
/* Use a buffer pool to minimize memcpy due to alignment problems. */
struct btr_pool *btrp;
};
/*
* If there is space left in the buffer of the stdin task, ask the scheduler to
* monitor STDIN_FILENO.
*/
static void stdin_pre_monitor(struct sched *s, void *context)
{
struct stdin_task *sit = context;
int ret;
ret = btr_node_status(sit->btrn, 0, BTR_NT_ROOT);
if (ret < 0)
sched_min_delay(s);
if (ret <= 0)
return;
if (btr_pool_unused(sit->btrp) > 0)
return sched_monitor_readfd(STDIN_FILENO, s);
sched_request_timeout_ms(100, s);
}
/*
* Feed data from stdin into the buffer tree if STDIN_FILENO is ready for
* reading.
*/
static int stdin_post_monitor(__a_unused struct sched *s, void *context)
{
struct stdin_task *sit = context;
ssize_t ret;
size_t sz, n;
char *buf = NULL;
ret = btr_node_status(sit->btrn, 0, BTR_NT_ROOT);
if (ret < 0)
goto err;
if (ret == 0)
return 0;
sz = btr_pool_get_buffer(sit->btrp, &buf);
if (sz == 0)
return 0;
if (!sched_read_ok(STDIN_FILENO, s))
return 0;
/*
* Do not use the maximal size to avoid having only a single buffer
* reference for the whole pool. This is bad because if that single
* reference can not be freed, we're stuck.
*/
sz = PARA_MIN(sz, btr_pool_size(sit->btrp) / 2);
ret = read_nonblock(STDIN_FILENO, buf, sz, &n);
if (n > 0)
btr_add_output_pool(sit->btrp, n, sit->btrn);
if (ret >= 0)
return 0;
err:
btr_remove_node(&sit->btrn);
return ret;
}
/**
* Allocate a stdin task structure and buffer tree node.
*
* \return An opaque pointer which identifies the newly created task.
* All errors are regarded as fatal, hence this function never returns NULL.
*/
struct stdin_task *stdin_new(void)
{
struct stdin_task *sit = alloc(sizeof(*sit));
sit->btrn = btr_new_node(&(struct btr_node_description) {
.name = "stdin"});
sit->btrp = btr_pool_new("stdin", 128 * 1024);
return sit;
}
/**
* Register an already allocated stdin task structure.
*
* \param sit As returned from \ref stdin_new().
* \param s The task will be added to this scheduler's task list.
*
* Tasks cannot be registered by the callers directly because only \ref stdin.c
* knows the static pre/post-monitor functions.
*
* \sa \ref stdout_register(), \ref sched_new().
*/
void stdin_register(struct stdin_task *sit, struct sched *s)
{
int ret;
task_register(&(struct task_info) {
.name = "stdin",
.pre_monitor = stdin_pre_monitor,
.post_monitor = stdin_post_monitor,
.context = sit,
}, s);
if (isatty(STDIN_FILENO))
return;
ret = mark_fd_nonblocking(STDIN_FILENO);
if (ret < 0) {
PARA_EMERG_LOG("set stdin to non-blocking mode: %s\n",
para_strerror(-ret));
exit(EXIT_FAILURE);
}
}
/**
* Get the buffer tree node of the stdin task.
*
* \param sit As returned from \ref stdin_new().
*
* \return A pointer to the node that was allocated in \ref stdin_new(). This
* never returns NULL.
*/
struct btr_node *stdin_btrn(const struct stdin_task *sit)
{
return sit->btrn;
}
/**
* Deallocate all resources allocated in \ref stdin_new().
*
* \param sit As returned from \ref stdin_new().
*
* This should be called after \ref schedule() has returned.
*/
void stdin_free(struct stdin_task *sit)
{
btr_pool_free(sit->btrp);
btr_remove_node(&sit->btrn);
free(sit);
}
|
