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
|
#pragma once
#include <Processors/IProcessor.h>
#include <queue>
namespace DB
{
/** Has arbitrary non zero number of inputs and arbitrary non zero number of outputs.
* All of them have the same structure.
*
* Pulls data from arbitrary input (whenever it is ready) and pushes it to arbitrary output (whenever it is not full).
* Doesn't do any heavy calculations.
* Doesn't preserve an order of data.
*
* Examples:
* - union data from multiple inputs to single output - to serialize data that was processed in parallel.
* - split data from single input to multiple outputs - to allow further parallel processing.
*/
class ResizeProcessor final : public IProcessor
{
public:
/// TODO Check that there is non zero number of inputs and outputs.
ResizeProcessor(const Block & header, size_t num_inputs, size_t num_outputs)
: IProcessor(InputPorts(num_inputs, header), OutputPorts(num_outputs, header))
, current_input(inputs.begin())
, current_output(outputs.begin())
{
}
String getName() const override { return "Resize"; }
Status prepare() override;
Status prepare(const PortNumbers &, const PortNumbers &) override;
private:
InputPorts::iterator current_input;
OutputPorts::iterator current_output;
size_t num_finished_inputs = 0;
size_t num_finished_outputs = 0;
std::queue<UInt64> waiting_outputs;
std::queue<UInt64> inputs_with_data;
bool initialized = false;
bool is_reading_started = false;
enum class OutputStatus
{
NotActive,
NeedData,
Finished,
};
enum class InputStatus
{
NotActive,
HasData,
Finished,
};
struct InputPortWithStatus
{
InputPort * port;
InputStatus status;
};
struct OutputPortWithStatus
{
OutputPort * port;
OutputStatus status;
};
std::vector<InputPortWithStatus> input_ports;
std::vector<OutputPortWithStatus> output_ports;
};
class StrictResizeProcessor : public IProcessor
{
public:
/// TODO Check that there is non zero number of inputs and outputs.
StrictResizeProcessor(const Block & header, size_t num_inputs, size_t num_outputs)
: IProcessor(InputPorts(num_inputs, header), OutputPorts(num_outputs, header))
, current_input(inputs.begin())
, current_output(outputs.begin())
{
}
StrictResizeProcessor(InputPorts inputs_, OutputPorts outputs_)
: IProcessor(inputs_, outputs_)
, current_input(inputs.begin())
, current_output(outputs.begin())
{
}
String getName() const override { return "StrictResize"; }
Status prepare(const PortNumbers &, const PortNumbers &) override;
private:
InputPorts::iterator current_input;
OutputPorts::iterator current_output;
size_t num_finished_inputs = 0;
size_t num_finished_outputs = 0;
std::queue<UInt64> disabled_input_ports;
std::queue<UInt64> waiting_outputs;
bool initialized = false;
enum class OutputStatus
{
NotActive,
NeedData,
Finished,
};
enum class InputStatus
{
NotActive,
NeedData,
Finished,
};
struct InputPortWithStatus
{
InputPort * port;
InputStatus status;
ssize_t waiting_output;
};
struct OutputPortWithStatus
{
OutputPort * port;
OutputStatus status;
};
std::vector<InputPortWithStatus> input_ports;
std::vector<OutputPortWithStatus> output_ports;
/// This field contained chunks which were read for output which had became finished while reading was happening.
/// They will be pushed to any next waiting output.
std::vector<Port::Data> abandoned_chunks;
};
}
|
