1 /*
2 * mptsd output writing
3 * Copyright (C) 2010-2011 Unix Solutions Ltd.
4 *
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License version 2
7 * as published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU General Public License for more details.
13 *
14 * You should have received a copy of the GNU General Public License
15 * along with this program; if not, write to the Free Software
16 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA.
17 */
18 #include <unistd.h>
19 #include <string.h>
20 #include <signal.h>
21 #include <sys/time.h>
22 #include <errno.h>
23 #include <math.h>
24
25 #include "libfuncs/io.h"
26 #include "libfuncs/log.h"
27 #include "libfuncs/list.h"
28
29 #include "libtsfuncs/tsfuncs.h"
30
31 #include "sleep.h"
32 #include "data.h"
33 #include "config.h"
34 #include "network.h"
35
increase_process_priority()36 void increase_process_priority() {
37 return;
38 #ifdef __linux__
39 struct sched_param param;
40 param.sched_priority = 99;
41 if (sched_setscheduler(0, SCHED_FIFO, ¶m)==-1) {
42 log_perror("sched_setscheduler() failed!", errno);
43 } else {
44 LOGf("PRIO : sched_setschedule() succeded.\n");
45 }
46 #endif
47 }
48
ts_frame_process(CONFIG * conf,OUTPUT * o,uint8_t * data)49 void ts_frame_process(CONFIG *conf, OUTPUT *o, uint8_t *data) {
50 int i;
51 uint16_t pid;
52 uint8_t *ts_packet;
53 for (i=0; i<FRAME_PACKET_SIZE; i+=TS_PACKET_SIZE) {
54 ts_packet = data + i;
55 pid = ts_packet_get_pid(ts_packet);
56
57 if (pid == 0x1fff) // NULL packet
58 o->padding_period += TS_PACKET_SIZE;
59
60 if (ts_packet_has_pcr(ts_packet)) {
61 uint64_t pcr = ts_packet_get_pcr(ts_packet); // Current PCR
62 uint64_t new_pcr = pcr;
63 uint64_t bytes = o->traffic + i;
64
65 if (o->last_pcr[pid]) {
66 uint64_t old_pcr = o->last_pcr[pid];
67 uint64_t old_org_pcr = o->last_org_pcr[pid];
68 uint64_t old_bytes = o->last_traffic[pid];
69 if (old_org_pcr < pcr) { // Detect PCR wraparound
70 new_pcr = old_pcr + (double)((bytes - old_bytes) * 8 * 27000000) / o->output_bitrate;
71 // Rewrite pcrs || Move pcrs & rewrite prcs
72 if (conf->pcr_mode == 2 || conf->pcr_mode == 3) {
73 ts_packet_set_pcr(ts_packet, new_pcr);
74 }
75 if (conf->debug) {
76 uint64_t ts_rate = (double)(((bytes - old_bytes) * 8) * 27000000) / (pcr - old_org_pcr);
77 uint64_t ts_rate_new = (double)(((bytes - old_bytes) * 8) * 27000000) / (new_pcr - old_pcr);
78 LOGf("PCR[%03x]: old:%14llu new:%14llu pcr_diff:%8lld ts_rate:%9llu ts_rate_new:%9llu diff:%9lld | passed:%llu\n",
79 pid,
80 pcr,
81 new_pcr,
82 pcr - new_pcr,
83 ts_rate,
84 ts_rate_new,
85 ts_rate - ts_rate_new,
86 bytes - old_bytes
87 );
88 }
89 }
90 } else {
91 // if (config->debug) {
92 // LOGf("PCR[%03x]: %10llu init\n", pid, pcr);
93 // }
94 }
95 o->last_pcr[pid] = new_pcr;
96 o->last_org_pcr[pid] = pcr;
97 o->last_traffic[pid] = bytes;
98 }
99 }
100 }
101
ts_frame_write(OUTPUT * o,uint8_t * data)102 ssize_t ts_frame_write(OUTPUT *o, uint8_t *data) {
103 ssize_t written;
104 written = fdwrite(o->out_sock, (char *)data, FRAME_PACKET_SIZE);
105 if (written >= 0) {
106 o->traffic += written;
107 o->traffic_period += written;
108 }
109
110 if (o->ofd)
111 write(o->ofd, data, FRAME_PACKET_SIZE);
112
113 return written;
114 }
115
output_handle_write(void * _config)116 void * output_handle_write(void *_config) {
117 CONFIG *conf = _config;
118 OUTPUT *o = conf->output;
119 int buf_in_use = 0;
120 unsigned int o_datasize = 0;
121 struct timeval stats_ts, now;
122 struct timeval start_write_ts, end_write_ts, used_ts;
123 unsigned long long stats_interval;
124
125 signal(SIGPIPE, SIG_IGN);
126
127 increase_process_priority();
128
129 gettimeofday(&stats_ts, NULL);
130 while (!o->dienow) {
131 gettimeofday(&now, NULL);
132 OBUF *curbuf = &o->obuf[buf_in_use];
133
134 while (curbuf->status != obuf_full) { // Wait untill the buffer is ready ot it is already emptying
135 if (o->dienow)
136 goto OUT;
137 //LOGf("MIX: Waiting for obuf %d\n", buf_in_use);
138 usleep(1);
139 }
140 curbuf->status = obuf_emptying; // Mark buffer as being filled
141
142 // Show stats
143 stats_interval = timeval_diff_msec(&stats_ts, &now);
144 if (stats_interval > conf->timeouts.stats) {
145 stats_ts = now;
146 double out_kbps = (double)(o->traffic_period * 8) / 1000;
147 double out_mbps = (double)out_kbps / 1000;
148 double opadding = ((double)o->padding_period / o->traffic_period) * 100;
149
150 if (!conf->quiet) {
151 LOGf("STAT : Pad:%6.2f%% Traf:%5.2f Mbps | %8.2f | %7llu\n",
152 opadding,
153 out_mbps,
154 out_kbps,
155 o->traffic_period
156 );
157 }
158 o->traffic_period = 0;
159 o->padding_period = 0;
160 o_datasize = 0;
161 }
162
163 gettimeofday(&start_write_ts, NULL);
164 int packets_written = 0, real_sleep_time = conf->output_tmout - conf->usleep_overhead;
165 long time_taken, time_diff, real_time, overhead = 0, overhead_total = 0;
166 ssize_t written = 0;
167 while (curbuf->written < curbuf->size) {
168 if (o->dienow)
169 goto OUT;
170 long sleep_interval = conf->output_tmout;
171 uint8_t *ts_frame = curbuf->buf + curbuf->written;
172 ts_frame_process(conf, o, ts_frame); // Fix PCR and count NULL packets
173 written += ts_frame_write(o, ts_frame); // Write packet to network/file
174 curbuf->written += FRAME_PACKET_SIZE;
175 if (packets_written) {
176 time_taken = timeval_diff_usec(&start_write_ts, &used_ts);
177 real_time = packets_written * (conf->output_tmout + conf->usleep_overhead);
178 time_diff = real_time - time_taken;
179 overhead = (time_taken / packets_written) - sleep_interval;
180 overhead_total += overhead;
181 /*
182 LOGf("[%5d] time_taken:%5ld real_time:%5ld time_diff:%ld | overhead:%5ld overhead_total:%5ld\n",
183 packets_written,
184 time_taken,
185 real_time,
186 time_diff,
187 overhead,
188 overhead_total
189 );
190 */
191 if (time_diff > real_sleep_time) {
192 sleep_interval = time_diff - conf->usleep_overhead;
193 if (sleep_interval < 0)
194 sleep_interval = 1;
195 // LOGf("Add sleep. time_diff: %ld sleep_interval: %ld\n", time_diff, sleep_interval);
196 } else {
197 //LOGf("Skip sleep %ld\n", time_diff);
198 sleep_interval = 0;
199 }
200
201 }
202 if (sleep_interval > 0)
203 usleep(sleep_interval);
204 gettimeofday(&used_ts, NULL);
205 packets_written++;
206 }
207 gettimeofday(&end_write_ts, NULL);
208 unsigned long long write_time = timeval_diff_usec(&start_write_ts, &end_write_ts);
209 if (write_time < o->obuf_ms * 1000) {
210 //LOGf("Writen for -%llu us less\n", o->obuf_ms*1000 - write_time);
211 usleep(o->obuf_ms*1000 - write_time);
212 } else {
213 //LOGf("Writen for +%llu us more\n", write_time - o->obuf_ms*1000);
214 }
215
216 obuf_reset(curbuf); // Buffer us all used up
217 buf_in_use = buf_in_use ? 0 : 1; // Switch buffer
218 if (written < 0) {
219 LOG("OUTPUT: Error writing into output socket.\n");
220 shutdown_fd(&o->out_sock);
221 connect_output(o);
222 }
223 }
224 OUT:
225 LOG("OUTPUT: WRITE thread stopped.\n");
226 o->dienow++;
227 return 0;
228 }
229