MythTV  master
mythfifowriter.cpp
Go to the documentation of this file.
1 // MythTV
2 #include "compat.h"
3 #include "mythlogging.h"
4 #include "mythconfig.h"
5 #if CONFIG_DARWIN
6 #include <sys/aio.h>
7 #endif
8 #include "io/mythfifowriter.h"
9 
10 // Std
11 #include <cstdio>
12 #include <cstdlib>
13 #include <unistd.h>
14 #include <fcntl.h>
15 #include <cassert>
16 #include <cerrno>
17 #include <sys/time.h>
18 #include <sys/types.h>
19 #include <sys/stat.h>
20 #include <ctime>
21 #include <cmath>
22 #include <iostream>
23 
25  : MThread("FIFOThread")
26 {
27 }
28 
30 {
31  wait();
32  m_parent = nullptr;
33  m_id = -1;
34 }
35 
37 {
38  m_id = Id;
39 }
40 
42 {
43  m_parent = Parent;
44 }
45 
47 {
48  RunProlog();
49  if (m_parent && m_id != -1)
51  RunEpilog();
52 }
53 
55  : m_numFifos(Count),
56  m_useSync(Sync)
57 {
58  if (Count < 1)
59  return;
60 
61  m_fifoBuf = new MythFifoBuffer*[Count];
62  m_fbInptr = new MythFifoBuffer*[Count];
63  m_fbOutptr = new MythFifoBuffer*[Count];
64  m_fifoThrds = new MythFIFOThread[Count];
65  m_fifoLock = new QMutex[Count];
66  m_fullCond = new QWaitCondition[Count];
67  m_emptyCond = new QWaitCondition[Count];
68  m_filename = new QString [Count];
69  m_fbDesc = new QString [Count];
70  m_maxBlkSize = new long[Count];
71  m_killWr = new int[Count];
72  m_fbCount = new int[Count];
73  m_fbMaxCount = new int[Count];
74 }
75 
77 {
78  if (m_numFifos < 1)
79  return;
80 
81  for (uint i = 0; i < m_numFifos; i++)
82  {
83  QMutexLocker flock(&m_fifoLock[i]);
84  m_killWr[i] = 1;
85  m_emptyCond[i].wakeAll();
86  }
87 
88  for (uint i = 0; i < m_numFifos; i++)
89  m_fifoThrds[i].wait();
90 
91  m_numFifos = 0;
92 
93  delete [] m_maxBlkSize;
94  delete [] m_fifoBuf;
95  delete [] m_fbInptr;
96  delete [] m_fbOutptr;
97  delete [] m_fifoThrds;
98  delete [] m_fullCond;
99  delete [] m_emptyCond;
100  delete [] m_fifoLock;
101  delete [] m_filename;
102  delete [] m_fbDesc;
103  delete [] m_killWr;
104  delete [] m_fbCount;
105  delete [] m_fbMaxCount;
106 }
107 
108 bool MythFIFOWriter::FIFOInit(uint Id, const QString& Desc, const QString& Name,
109  long Size, int NumBufs)
110 {
111  if (Id >= m_numFifos)
112  return false;
113 
114  QByteArray fname = Name.toLatin1();
115  const char *aname = fname.constData();
116  if (mkfifo(aname, S_IREAD | S_IWRITE | S_IRGRP | S_IROTH) == -1)
117  {
118  LOG(VB_GENERAL, LOG_ERR, QString("Couldn't create fifo for file: '%1'").arg(Name) + ENO);
119  return false;
120  }
121 
122  LOG(VB_GENERAL, LOG_INFO, QString("Created %1 fifo: %2").arg(Desc).arg(Name));
123 
124  m_maxBlkSize[Id] = Size;
125  m_filename[Id] = Name;
126  m_fbDesc[Id] = Desc;
127  m_killWr[Id] = 0;
128  m_fbCount[Id] = (m_useSync) ? 2 : NumBufs;
129  m_fbMaxCount[Id] = 512;
130  m_fifoBuf[Id] = new MythFifoBuffer;
131  struct MythFifoBuffer *fifoptr = m_fifoBuf[Id];
132  for (int i = 0; i < m_fbCount[Id]; i++)
133  {
134  fifoptr->m_data = new unsigned char[static_cast<unsigned long>(m_maxBlkSize[Id])];
135  if (i == m_fbCount[Id] - 1)
136  fifoptr->m_next = m_fifoBuf[Id];
137  else
138  fifoptr->m_next = new struct MythFifoBuffer;
139  fifoptr = fifoptr->m_next;
140  }
141  m_fbInptr[Id] = m_fifoBuf[Id];
142  m_fbOutptr[Id] = m_fifoBuf[Id];
143 
144  m_fifoThrds[Id].SetParent(this);
145  m_fifoThrds[Id].SetId(static_cast<int>(Id));
146  m_fifoThrds[Id].start();
147 
148  while (0 == m_killWr[Id] && !m_fifoThrds[Id].isRunning())
149  usleep(1000);
150 
151  return m_fifoThrds[Id].isRunning();
152 }
153 
155 {
156  int fd = -1;
157 
158  QMutexLocker flock(&m_fifoLock[Id]);
159  while (true)
160  {
161  if ((m_fbInptr[Id] == m_fbOutptr[Id]) && (0 == m_killWr[Id]))
162  m_emptyCond[Id].wait(flock.mutex());
163  flock.unlock();
164  if (m_killWr[Id])
165  break;
166  if (fd < 0)
167  {
168  QByteArray fname = m_filename[Id].toLatin1();
169  fd = open(fname.constData(), O_WRONLY| O_SYNC);
170  }
171  if (fd >= 0)
172  {
173  int written = 0;
174  while (written < m_fbOutptr[Id]->m_blockSize)
175  {
176  int ret = static_cast<int>(write(fd, m_fbOutptr[Id]->m_data + written,
177  static_cast<size_t>(m_fbOutptr[Id]->m_blockSize-written)));
178  if (ret < 0)
179  {
180  LOG(VB_GENERAL, LOG_ERR, QString("FIFOW: write failed with %1")
181  .arg(strerror(errno)));
183  break;
184  }
185  written += ret;
186  }
187  }
188  flock.relock();
189  m_fbOutptr[Id] = m_fbOutptr[Id]->m_next;
190  m_fullCond[Id].wakeAll();
191  }
192 
193  if (fd != -1)
194  close(fd);
195 
196  unlink(m_filename[Id].toLocal8Bit().constData());
197 
198  while (m_fifoBuf[Id]->m_next != m_fifoBuf[Id])
199  {
200  struct MythFifoBuffer *tmpfifo = m_fifoBuf[Id]->m_next->m_next;
201  delete [] m_fifoBuf[Id]->m_next->m_data;
202  delete m_fifoBuf[Id]->m_next;
203  m_fifoBuf[Id]->m_next = tmpfifo;
204  }
205  delete [] m_fifoBuf[Id]->m_data;
206  delete m_fifoBuf[Id];
207 }
208 
209 void MythFIFOWriter::FIFOWrite(uint Id, void *Buffer, long Size)
210 {
211  QMutexLocker flock(&m_fifoLock[Id]);
212  while (m_fbInptr[Id]->m_next == m_fbOutptr[Id])
213  {
214  bool blocking = false;
215  if (!m_useSync)
216  {
217  for (uint i = 0; i < m_numFifos; i++)
218  {
219  if (i == Id)
220  continue;
221  if (m_fbInptr[i] == m_fbOutptr[i])
222  blocking = true;
223  }
224  }
225 
226  if (blocking && m_fbCount[Id] < m_fbMaxCount[Id])
227  {
228  struct MythFifoBuffer *tmpfifo = m_fbInptr[Id]->m_next;
229  m_fbInptr[Id]->m_next = new struct MythFifoBuffer;
230  m_fbInptr[Id]->m_next->m_data = new unsigned char[static_cast<unsigned long>(m_maxBlkSize[Id])];
231  m_fbInptr[Id]->m_next->m_next = tmpfifo;
232  QString msg = QString("allocating additonal buffer for : %1(%2)")
233  .arg(m_fbDesc[Id]).arg(++m_fbCount[Id]);
234  LOG(VB_FILE, LOG_INFO, msg);
235  }
236  else
237  {
238  m_fullCond[Id].wait(flock.mutex(), 1000);
239  }
240  }
241 
242  if (Size > m_maxBlkSize[Id])
243  {
244  delete [] m_fbInptr[Id]->m_data;
245  m_fbInptr[Id]->m_data = new unsigned char[static_cast<unsigned long>(Size)];
246  }
247 
248  memcpy(m_fbInptr[Id]->m_data,Buffer, static_cast<size_t>(Size));
249  m_fbInptr[Id]->m_blockSize = Size;
250  m_fbInptr[Id] = m_fbInptr[Id]->m_next;
251  m_emptyCond[Id].wakeAll();
252 }
253 
255 {
256  uint count = 0;
257  while (count < m_numFifos)
258  {
259  count = 0;
260  for (uint i = 0; i < m_numFifos; i++)
261  {
262  QMutexLocker flock(&m_fifoLock[i]);
263  if (m_fbInptr[i] == m_fbOutptr[i])
264  {
265  m_killWr[i] = 1;
266  m_emptyCond[i].wakeAll();
267  count++;
268  }
269  }
270  usleep(1000);
271  }
272 }
O_SYNC
#define O_SYNC
Definition: compat.h:226
MThread::start
void start(QThread::Priority p=QThread::InheritPriority)
Tell MThread to start running the thread in the near future.
Definition: mthread.cpp:286
ENO
#define ENO
This can be appended to the LOG args with "+".
Definition: mythlogging.h:72
MythFIFOWriter::m_fbOutptr
MythFifoBuffer ** m_fbOutptr
Definition: mythfifowriter.h:56
MythFIFOWriter::m_maxBlkSize
long * m_maxBlkSize
Definition: mythfifowriter.h:66
MythFIFOWriter::m_emptyCond
QWaitCondition * m_emptyCond
Definition: mythfifowriter.h:61
S_IROTH
#define S_IROTH
Definition: compat.h:225
MythFIFOWriter::m_fbDesc
QString * m_fbDesc
Definition: mythfifowriter.h:64
MThread::wait
bool wait(std::chrono::milliseconds time=std::chrono::milliseconds::max())
Wait for the MThread to exit, with a maximum timeout.
Definition: mthread.cpp:303
MythFIFOThread::~MythFIFOThread
~MythFIFOThread() override
Definition: mythfifowriter.cpp:29
mythburn.write
def write(text, progress=True)
Definition: mythburn.py:308
MythFIFOThread::MythFIFOThread
MythFIFOThread()
Definition: mythfifowriter.cpp:24
arg
arg(title).arg(filename).arg(doDelete))
MythFIFOWriter::~MythFIFOWriter
~MythFIFOWriter(void)
Definition: mythfifowriter.cpp:76
MythFIFOWriter::m_fullCond
QWaitCondition * m_fullCond
Definition: mythfifowriter.h:60
MythFIFOThread
Definition: mythfifowriter.h:15
LOG
#define LOG(_MASK_, _LEVEL_, _QSTRING_)
Definition: mythlogging.h:23
MThread::RunProlog
void RunProlog(void)
Sets up a thread, call this if you reimplement run().
Definition: mthread.cpp:196
MythFIFOWriter::m_useSync
bool m_useSync
Definition: mythfifowriter.h:71
MythFIFOWriter
Definition: mythfifowriter.h:31
MythFIFOThread::m_id
int m_id
Definition: mythfifowriter.h:28
MythFIFOWriter::m_fbInptr
MythFifoBuffer ** m_fbInptr
Definition: mythfifowriter.h:55
MythFIFOWriter::FIFODrain
void FIFODrain(void)
Definition: mythfifowriter.cpp:254
MythFIFOThread::SetParent
void SetParent(MythFIFOWriter *Parent)
Definition: mythfifowriter.cpp:41
close
#define close
Definition: compat.h:17
MythFIFOWriter::MythFifoBuffer
Definition: mythfifowriter.h:47
MythFIFOWriter::MythFifoBuffer::m_blockSize
long m_blockSize
Definition: mythfifowriter.h:51
mythlogging.h
compat.h
MythFIFOThread::run
void run(void) override
Runs the Qt event loop unless we have a QRunnable, in which case we run the runnable run instead.
Definition: mythfifowriter.cpp:46
MythFIFOWriter::FIFOInit
bool FIFOInit(uint Id, const QString &Desc, const QString &Name, long Size, int NumBufs)
Definition: mythfifowriter.cpp:108
mythfifowriter.h
MThread::RunEpilog
void RunEpilog(void)
Cleans up a thread's resources, call this if you reimplement run().
Definition: mthread.cpp:209
S_IRGRP
#define S_IRGRP
Definition: compat.h:224
uint
unsigned int uint
Definition: compat.h:140
MythFIFOWriter::MythFifoBuffer::m_next
struct MythFifoBuffer * m_next
Definition: mythfifowriter.h:49
MythFIFOWriter::FIFOWrite
void FIFOWrite(uint Id, void *Buffer, long Size)
Definition: mythfifowriter.cpp:209
MythFIFOWriter::m_filename
QString * m_filename
Definition: mythfifowriter.h:63
MythFIFOWriter::m_fifoLock
QMutex * m_fifoLock
Definition: mythfifowriter.h:59
MythFIFOWriter::m_fbCount
int * m_fbCount
Definition: mythfifowriter.h:68
MythFIFOWriter::MythFifoBuffer::m_data
unsigned char * m_data
Definition: mythfifowriter.h:50
isRunning
static bool isRunning(const char *program)
Returns true if a program containing the specified string is running on this machine.
Definition: mythtv/programs/mythshutdown/main.cpp:202
MythFIFOWriter::m_fbMaxCount
int * m_fbMaxCount
Definition: mythfifowriter.h:69
Buffer
Definition: MythExternControl.h:36
Name
Definition: channelsettings.cpp:46
MythFIFOWriter::FIFOWriteThread
void FIFOWriteThread(int Id)
Definition: mythfifowriter.cpp:154
MythFIFOWriter::m_numFifos
uint m_numFifos
Definition: mythfifowriter.h:70
MThread
This is a wrapper around QThread that does several additional things.
Definition: mthread.h:49
MThread::isRunning
bool isRunning(void) const
Definition: mthread.cpp:266
MythFIFOWriter::m_killWr
int * m_killWr
Definition: mythfifowriter.h:67
MythFIFOThread::m_parent
MythFIFOWriter * m_parent
Definition: mythfifowriter.h:27
MythFIFOWriter::MythFIFOWriter
MythFIFOWriter(uint Count, bool Sync)
Definition: mythfifowriter.cpp:54
MythFIFOWriter::m_fifoThrds
MythFIFOThread * m_fifoThrds
Definition: mythfifowriter.h:58
MythFIFOThread::SetId
void SetId(int Id)
Definition: mythfifowriter.cpp:36
MythFIFOWriter::m_fifoBuf
MythFifoBuffer ** m_fifoBuf
Definition: mythfifowriter.h:54
mkfifo
#define mkfifo(path, mode)
Definition: compat.h:228