]> git.sesse.net Git - casparcg/blob - core/producer/channel/channel_producer.cpp
Fixed bug in channel_producer
[casparcg] / core / producer / channel / channel_producer.cpp
1 /*\r
2 * Copyright 2013 Sveriges Television AB http://casparcg.com/\r
3 *\r
4 * This file is part of CasparCG (www.casparcg.com).\r
5 *\r
6 * CasparCG is free software: you can redistribute it and/or modify\r
7 * it under the terms of the GNU General Public License as published by\r
8 * the Free Software Foundation, either version 3 of the License, or\r
9 * (at your option) any later version.\r
10 *\r
11 * CasparCG is distributed in the hope that it will be useful,\r
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of\r
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\r
14 * GNU General Public License for more details.\r
15 *\r
16 * You should have received a copy of the GNU General Public License\r
17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.\r
18 *\r
19 * Author: Robert Nagy, ronag89@gmail.com\r
20 */\r
21 \r
22 #include "../../stdafx.h"\r
23 \r
24 #include "channel_producer.h"\r
25 \r
26 #include "../../monitor/monitor.h"\r
27 #include "../../consumer/frame_consumer.h"\r
28 #include "../../consumer/output.h"\r
29 #include "../../video_channel.h"\r
30 \r
31 #include "../frame/basic_frame.h"\r
32 #include "../frame/frame_factory.h"\r
33 #include "../../mixer/write_frame.h"\r
34 #include "../../mixer/read_frame.h"\r
35 \r
36 #include <common/exception/exceptions.h>\r
37 #include <common/memory/memcpy.h>\r
38 #include <common/concurrency/future_util.h>\r
39 \r
40 #include <tbb/concurrent_queue.h>\r
41 \r
42 namespace caspar { namespace core {\r
43 \r
44 class channel_consumer : public frame_consumer\r
45 {       \r
46         tbb::concurrent_bounded_queue<std::shared_ptr<read_frame>>      frame_buffer_;\r
47         core::video_format_desc                                                                         format_desc_;\r
48         int                                                                                                                     channel_index_;\r
49         tbb::atomic<bool>                                                                                       is_running_;\r
50         tbb::atomic<int64_t>                                                                            current_age_;\r
51 \r
52 public:\r
53         channel_consumer() \r
54         {\r
55                 is_running_ = true;\r
56                 current_age_ = 0;\r
57                 frame_buffer_.set_capacity(3);\r
58         }\r
59 \r
60         ~channel_consumer()\r
61         {\r
62                 stop();\r
63         }\r
64 \r
65         // frame_consumer\r
66 \r
67         virtual boost::unique_future<bool> send(const safe_ptr<read_frame>& frame) override\r
68         {\r
69                 frame_buffer_.try_push(frame);\r
70                 return caspar::wrap_as_future(is_running_.load());\r
71         }\r
72 \r
73         virtual void initialize(const core::video_format_desc& format_desc, int channel_index) override\r
74         {\r
75                 format_desc_    = format_desc;\r
76                 channel_index_  = channel_index;\r
77         }\r
78 \r
79         virtual int64_t presentation_frame_age_millis() const override\r
80         {\r
81                 return current_age_;\r
82         }\r
83 \r
84         virtual std::wstring print() const override\r
85         {\r
86                 return L"[channel-consumer|" + boost::lexical_cast<std::wstring>(channel_index_) + L"]";\r
87         }\r
88 \r
89         virtual boost::property_tree::wptree info() const override\r
90         {\r
91                 boost::property_tree::wptree info;\r
92                 info.add(L"type", L"channel-consumer");\r
93                 info.add(L"channel-index", channel_index_);\r
94                 return info;\r
95         }\r
96         \r
97         virtual bool has_synchronization_clock() const override\r
98         {\r
99                 return false;\r
100         }\r
101 \r
102         virtual size_t buffer_depth() const override\r
103         {\r
104                 return 1;\r
105         }\r
106 \r
107         virtual int index() const override\r
108         {\r
109                 return 78500 + channel_index_;\r
110         }\r
111 \r
112         // channel_consumer\r
113 \r
114         void stop()\r
115         {\r
116                 is_running_ = false;\r
117                 frame_buffer_.try_push(make_safe<read_frame>());\r
118         }\r
119         \r
120         const core::video_format_desc& get_video_format_desc()\r
121         {\r
122                 return format_desc_;\r
123         }\r
124 \r
125         std::shared_ptr<read_frame> receive()\r
126         {\r
127                 if(!is_running_)\r
128                         return make_safe<read_frame>();\r
129                 std::shared_ptr<read_frame> frame;\r
130                 \r
131                 if (frame_buffer_.try_pop(frame))\r
132                         current_age_ = frame->get_age_millis();\r
133 \r
134                 return frame;\r
135         }\r
136 };\r
137         \r
138 class channel_producer : public frame_producer\r
139 {\r
140         monitor::subject                                        monitor_subject_;\r
141 \r
142         const safe_ptr<frame_factory>           frame_factory_;\r
143         const safe_ptr<channel_consumer>        consumer_;\r
144 \r
145         std::queue<safe_ptr<basic_frame>>       frame_buffer_;\r
146         safe_ptr<basic_frame>                           last_frame_;\r
147         uint64_t                                                        frame_number_;\r
148 \r
149 public:\r
150         explicit channel_producer(const safe_ptr<frame_factory>& frame_factory, const safe_ptr<video_channel>& channel) \r
151                 : frame_factory_(frame_factory)\r
152                 , consumer_(make_safe<channel_consumer>())\r
153                 , last_frame_(basic_frame::empty())\r
154                 , frame_number_(0)\r
155         {\r
156                 channel->output()->add(consumer_);\r
157                 CASPAR_LOG(info) << print() << L" Initialized";\r
158         }\r
159 \r
160         ~channel_producer()\r
161         {\r
162                 consumer_->stop();\r
163                 CASPAR_LOG(info) << print() << L" Uninitialized";\r
164         }\r
165 \r
166         // frame_producer\r
167                         \r
168         virtual safe_ptr<basic_frame> receive(int) override\r
169         {\r
170                 auto format_desc = consumer_->get_video_format_desc();\r
171 \r
172                 if(frame_buffer_.size() > 1)\r
173                 {\r
174                         auto frame = frame_buffer_.front();\r
175                         frame_buffer_.pop();\r
176                         return last_frame_ = frame;\r
177                 }\r
178                 \r
179                 auto read_frame = consumer_->receive();\r
180                 if(!read_frame || read_frame->image_data().empty())\r
181                         return basic_frame::late();             \r
182 \r
183                 frame_number_++;\r
184                 \r
185                 core::pixel_format_desc desc;\r
186                 bool double_speed       = std::abs(frame_factory_->get_video_format_desc().fps / 2.0 - format_desc.fps) < 0.01;         \r
187                 bool half_speed         = std::abs(format_desc.fps / 2.0 - frame_factory_->get_video_format_desc().fps) < 0.01;\r
188 \r
189                 if(half_speed && frame_number_ % 2 == 0) // Skip frame\r
190                         return receive(0);\r
191 \r
192                 desc.pix_fmt = core::pixel_format::bgra;\r
193                 desc.planes.push_back(core::pixel_format_desc::plane(format_desc.width, format_desc.height, 4));\r
194                 auto frame = frame_factory_->create_frame(this, desc);\r
195 \r
196                 fast_memcpy(frame->image_data().begin(), read_frame->image_data().begin(), read_frame->image_data().size());\r
197                 frame->commit();\r
198 \r
199                 frame_buffer_.push(frame);      \r
200                 \r
201                 if(double_speed)        \r
202                         frame_buffer_.push(frame);\r
203 \r
204                 return receive(0);\r
205         }       \r
206 \r
207         virtual safe_ptr<basic_frame> last_frame() const override\r
208         {\r
209                 return last_frame_; \r
210         }       \r
211 \r
212         virtual std::wstring print() const override\r
213         {\r
214                 return L"channel[]";\r
215         }\r
216 \r
217         virtual boost::property_tree::wptree info() const override\r
218         {\r
219                 boost::property_tree::wptree info;\r
220                 info.add(L"type", L"channel-producer");\r
221                 return info;\r
222         }\r
223 \r
224         monitor::source& monitor_output() \r
225         {\r
226                 return monitor_subject_;\r
227         }\r
228 };\r
229 \r
230 safe_ptr<frame_producer> create_channel_producer(const safe_ptr<core::frame_factory>& frame_factory, const safe_ptr<video_channel>& channel)\r
231 {\r
232         return create_producer_print_proxy(\r
233                         make_safe<channel_producer>(frame_factory, channel));\r
234 }\r
235 \r
236 }}