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
|
/* SPDX-License-Identifier: GPL-2.0 */
/** \file stdout.c Functions that deal with writing to stdout. */
#include "para.h"
#include "list.h"
#include "sched.h"
#include "fd.h"
#include "error.h"
#include "string.h"
#include "buffer_tree.h"
#include "stdout.h"
/* para_recv, para_filter and para_client create an stdout task. */
struct stdout_task {
/* Stdout is always a leaf node of the buffer tree. */
struct btr_node *btrn;
};
/* Monitor STDOUT_FILENO if there is input data available. */
static void stdout_pre_monitor(struct sched *s, void *context)
{
struct stdout_task *sot = context;
int ret;
ret = btr_node_status(sot->btrn, 0, BTR_NT_LEAF);
if (ret > 0)
sched_monitor_writefd(STDOUT_FILENO, s);
else if (ret < 0)
sched_min_delay(s);
}
/*
* If input from the buffer tree is available and STDOUT_FILENO is ready, write
* as much as possible.
*/
static int stdout_post_monitor(struct sched *s, void *context)
{
struct stdout_task *sot = context;
struct btr_node *btrn = sot->btrn;
int ret;
char *buf;
size_t sz;
ret = btr_node_status(btrn, 0, BTR_NT_LEAF);
if (ret < 0)
goto out;
if (ret == 0)
return 0;
if (!sched_write_ok(STDOUT_FILENO, s))
return 0;
for (;;) {
sz = btr_next_buffer(btrn, &buf);
if (sz == 0)
break;
ret = xwrite(STDOUT_FILENO, buf, sz);
if (ret <= 0)
break;
btr_consume(btrn, ret);
}
out:
if (ret < 0)
btr_remove_node(&sot->btrn);
return ret;
}
/**
* Register a stdout task structure.
*
* \param parent From where we receive the data for stdout.
*
* \return An opaque pointer which identifies the newly created task.
* All errors are regarded as fatal, hence this function never returns NULL.
*/
struct stdout_task *stdout_new(struct btr_node *parent)
{
struct stdout_task *sot = zalloc(sizeof(*sot));
sot->btrn = btr_new_node(&(struct btr_node_description)
{.name = "stdout", .parent = parent});
return sot;
}
/**
* Register an instance of the stdout task to the scheduler.
*
* \param sot As returned by \ref stdout_new().
* \param s The task will be added to this scheduler's task list.
*
* \sa \ref stdin_register(), \ref sched_new().
*/
void stdout_register(struct stdout_task *sot, struct sched *s)
{
int ret;
task_register(&(struct task_info) {
.name = "stdout",
.pre_monitor = stdout_pre_monitor,
.post_monitor = stdout_post_monitor,
.context = sot,
}, s);
if (isatty(STDOUT_FILENO))
return;
ret = mark_fd_nonblocking(STDOUT_FILENO);
if (ret < 0) {
PARA_EMERG_LOG("set stdout to non-blocking mode: %s\n",
para_strerror(-ret));
exit(EXIT_FAILURE);
}
}
/**
* Get the buffer tree node of the stdout task.
*
* \param sot As returned from \ref stdout_new().
*
* \return A pointer to the node that was allocated in \ref stdout_new(). This
* never returns NULL.
*/
struct btr_node *stdout_btrn(const struct stdout_task *sot)
{
return sot->btrn;
}
/**
* Deallocate all resources allocated in \ref stdout_new().
*
* \param sot As returned from \ref stdout_new().
*
* This should be called after \ref schedule() has returned.
*/
void stdout_free(struct stdout_task *sot)
{
btr_remove_node(&sot->btrn);
free(sot);
}
|
