Introduction
Few programmers are lucky to have witnessed the birth of a project. Say, someone comes up with a brilliant idea such as to stream video over the network. Next thing you know, a design gets created, and C++ source code starts to appear on blank screens. This would be dream come true as you would be the expert that other programmers refer to many years down the road. Most of us, however, are relegated to work on baseline source code that has been around for many years. Even worse, some source code gets grandfathered to us from other groups of programmers. In this situation, sometimes, you do not really have a choice but to own other people's coding styles and pick their bad habits on the way. Assimilation means you just do not want your good code to stand out like a tree in a field of grass. But, just for today, we are going to pretend to be the guys and gals that the company has entrusted with initiating a coding project.
Our goal is to create an application where a video is played on a screen of one machine, and another connected machine is able to replicate that video in real-time. By the looks of it, we will need two processes—one to send and play video frames and another one to receive and play those frames. These two processes will need to communicate over a network protocol—a reliable one. Since we intend on sending video frames over the network, the order in which they get received is crucial. You can imagine the downgrade to video quality if some of the frames get eaten up (or lost) by the network. TCP is a connection-oriented protocol which means data transfer will not start until a reliable connection is established. In this protocol, all data packets arrive destination in the order they were sent because they follow the same route. If the route cannot be maintained, another route maybe established and the destination node will demand that all the lost packets be resent before it can process new ones. TCP is, then, exactly what we need for this application—the video will continue from where it was stopped before the technical difficulty with the route occurred.
We know that scripting and programming languages are the tools of trade when it comes to the making of an application. The choice of the right tool is of paramount importance in the accomplishment of tasks for a job well done. You simply cannot bring down a large tree with a small wrench; on the other hand, it is impossible to unscrew a tight bolt by using a saw. However, if given the choice, try to stay away from Visual Studio languages (C#, Visual Basic, Visual C++) unless you do not mind being tethered to the Windows platform. The choice amongst popular scripting and programming languages now narrows down to Java, Python, or C++. Java interpreter and JVM may get in the way of fast rendering of the video frames, and with Python, we might not get finesse in controlling execution of the program. For example, we would be out of luck if we are not satisfied with memory performance of a dictionary container in Python. Of-course, one can always improve any feature of Pythonby going to the source code, which leads us to the final choice. The obvious choice is then C++ with its close to the metal deal. I just got to tell you there is nothing wrong with a good choice if you are feeling led to the choice of C++.
Now, let us take a look at the means to our disposal for displaying video. If you have ever watched a video on YouTube, then you should assure yourself that there is way to transfer and display video for your application as well. It is called video streaming. A quick Googling on the subject provides with lots of hits for OpenCV. This is an open source library with algorithms for image processing. It has interfaces for C++, C, and Python. The library contains functions for playing and manipulating images and by extension video. Download and install OpenCV. Now looks like we have all of our ducks in row.
Birthing the Processes
Like many of you, I was introduced to programming with a small program to display “Hello world!” I must say, at the time, I did not appreciate that program to any degree. For all I cared, I was always able to display on screen anything I typed over a text editor. I was showed to compile and link the program in a certain way, and that was how I created the executable and was able to display the dumb message “Hello world!” without putting much thought into what was happening under the hood. That was then when I thought there could not be anything dumber than echoing a message. Flash forward to now. Now, I find myself printing tons of messages while debugging when I find it hard to use a debugger for one reason or another. Displaying output is generally an acceptable way of checking if things are in working order. Here, too, we are going to follow suite.
Create two text files by the names VideoSender.cpp and VideoReceiver.cpp and enter the following text.
#include <iostream>
#include "process_metrics.h"
int senderMainEntry()
{
std::cout << "Video sender ..." << std::endl;
return 0;
}
int main()
{
return senderMainEntry();
}
#include <iostream>
#include "process_metrics.h"
int receiverMainEntry()
{
std::cout << "Video receiver ..." << std::endl;
return 0;
}
int main()
{
return receiverMainEntry();
}
The code in these two files is pretty much standard—not much is happening yet. C++ does not allow us to change the name of the entry point function from that of main(), but I found it useful, in the past, to have a uniquely named entry point function for each process. In the setting of a large project with a number of processes, it helps, when using the IDE's call hierarchy, to know in which process the call hierarchy originated. Thus, the purpose of the main() function here is to call the process' entry point function. And, at this point, we have brought into the world the two processes.
Turn your focus to something else for just a second. Next, a basic metrics tool will be included in both of the processes. Create a new file called process_metrics.h with this text.
#include <stdio.h>
#include <time.h>
#define LOCAL_TIME(x) \
time_t t = time(NULL); \
struct tm * locTime = localtime(&t); \
char x[9]; \
strftime(x, 9, "%T", locTime); \
while(0)
clock_t pTicks;
__attribute__((constructor)) void begin ()
{
LOCAL_TIME(sLocTime);
printf("begin execution: %s\n", sLocTime);
pTicks = clock();
}
__attribute__((destructor)) void end ()
{
pTicks = clock() - pTicks;
printf("Number of clock ticks: %ld\n", pTicks);
LOCAL_TIME(sLocTime);
printf("end execution: %s\n", sLocTime);
}
Some of you may find this code particularly strange. Especially, the definitions __attribute__((constructor)) void begin() and __attribute__((destructor)) void end() may look alien to normal C++. And, strangely enough, as you saw in videoSender.cpp and videoReceiver.cpp, I did not add invocations to begin() and end() as I totally expect them to run on their own. Well, what is happening is that I am adding begin() to the global constructor list (CTOR) and end() to the global destructor list (DTOR). Global objects must have constructor and destructor like any other object in C++, and, as such, all functions in CTOR and DTOR lists are automatically run. As far as the user is concerned, a process starts at main() and terminates at exit(0), but each function in CTOR is called before main() and all the functions in DTOR are called after exit(). The rest of the code is there to display start and end time for the process and the number of clock cycles it took.
Acquainting OpenCV
In the age of the Internet, the worst thing you can do is reinvent the wheel (well, other than all the nasty stuff you are able to do with the Internet). Chances are you are not the first one to have the problem you are trying to solve; better human beings have already written code for your problem and made the source code available with a free license. You can simply use their code or read it to learn from. Who knows, on the way, you might end up becoming the one to take it to the next level by contributing to that open source program.
Not knowing much about OpenCV, I would describe it as a package of shared libraries with functions for image processing. Using OpenCV, you are able to read frames off of a video, do processing, and display. But, OpenCV needs codecs to open video files. I downloaded a package containing most of common codecs out there and installed it on my Ubuntu VM.
> sudo apt-get install -y ubuntu-restricted-extras
Next, I downloaded, compiled and linked OpenCV (compiling and linking is what I call “building”). I had to go through a few iterations of the build options and debug OpenCV to try to make it work until, finally, I found an article by a Manuel Ignacio López Quintero that details the building steps. This is the final version of cmake options listing that yielded the result:
> mkdir build
>
> cd build
>
> cmake -DWITH_IPP=ON -DWITH_OPENGL=ON -DWITH_GTK=ON -DWITH_VTK=ON <code class="western">-DCMAKE_INSTALL_PREFIX=./build ..
>
> make install
OpenCV will not be installed in the standard directories of /usr/bin and /usr/lib but in ./build (see CMAKE_INSTALL_PREFIX). Since we are installing OpenCV in an unconventional manner, do not forget to run
> sudo ldconfig
It is now time to configure the project. But, before configuring the build, you must make sure you have the dependencies in place. For example, since you will need to display the video in a window, download and install the latest development libraries for GTK and VTK.
> sudo apt-get install -y libgtk-3-dev libvtk6-dev
While on the topic of GTK and GUI, OpenCV can also be configured to use Qt—the other GUI development library. And, when dealing with GUI programming, parallelism is a bull to be held by the horns; OpenCV can be configured with TBB to deal with multi-threading, but, otherwise, OpenCV uses pThreads.
Thus, at the end of it all, I was able to run a video file with this modification to videoSender.cpp.
#include <iostream>
#include "process_metrics.h"
#include <opencv2/opencv.hpp>
int senderMainEntry (const char* videoFile)
{
std::cout << "Video sender ..." << std::endl;
cv::VideoCapture capturedVideo;
capturedVideo.open(videoFile);
if (!capturedVideo.isOpened())
{
std::cerr << "Unable to open video: " << videoFile << std::endl;
return -1;
}
cv::Mat dispFrame;
cv::namedWindow("Video Send", cv::WINDOW_AUTOSIZE);
while (true)
{
capturedVideo >> dispFrame;
if (dispFrame.empty())
{
break;
}
else
{
imshow("Video Send", dispFrame);
}
if (cv::waitKey(30) == 0)
{
puts("Key press");
break;
}
}
return 0;
}
int main(int argv, char** argc)
{
return senderMainEntry(argc[1]);
}
As you can see, the code in videoSender.cpp was modified in a manner as to be able to read a video file and display it frame by frame.
Connecting and Signaling the Processes
Application level communication takes place between processes. The NIC, which has a unique IP address in the network, is shared by many ports that processes have access to it. A port is identified by a port number, and communicating processes must know each others port numbers. The operating system gives access of the port and IP address to the process by what is known as a socket. Here, you will find a class dubbed SocketHandler that abstracts the use of a socket.
#ifndef SOCKETHANDLER_H
#define SOCKETHANDLER_H
#include <functional>
#include <string>
#include <sys/types.h>
#include <netinet/in.h>
#include <unistd.h>
#define PORT_ONE 51001
#define PORT_TWO 51002
class SocketHandler
{
enum InitializationStatus
{
Uninitialized,
Initialized
};
const std::size_t PACKET_SIZE = 21845;
InitializationStatus init_status;
int socket_fd;
int connected_socket_fd;
int port_no;
struct sockaddr_in connect_to_addr;
bool is_client_connected = false;
public:
SocketHandler (int socket_type, int port);
~SocketHandler ();
int InitializeAsServer ();
void ConnectToServer (std::string ip_address);
void ConnectToClient ();
int GetFileDescriptor ();
ssize_t Send (const void * data, std::size_t len);
ssize_t Receive (void * buffer, std::size_t len = 1);
ssize_t SendLarge (const void * data, std::size_t len);
ssize_t ReceiveLarge (void * buffer, std::size_t len);
ssize_t SendAndWait (const void * data, std::size_t len);
ssize_t ReceiveAndAcknowledge (void * buffer, std::size_t len, std::function <void(void *)> PerformBeforeAcknowledging);
void CloseSocket ();
};
inline int SocketHandler::GetFileDescriptor ()
{
return is_client_connected ? connected_socket_fd : socket_fd;
}
inline ssize_t SocketHandler::Send (const void * data, std::size_t len)
{
return send(GetFileDescriptor(), data, len, 0);
}
inline ssize_t SocketHandler::Receive (void * buffer, std::size_t len)
{
return recv(GetFileDescriptor(), buffer, len, 0);
}
inline void SocketHandler::CloseSocket ()
{
close(GetFileDescriptor());
}
#endif
#include <sys/poll.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <strings.h>
#include "SocketHandler.h"
SocketHandler::SocketHandler (int socket_type, int port):
init_status(Uninitialized),
port_no(port)
{
socket_fd = socket(AF_INET, socket_type, 0);
if (socket_fd < 0)
{
perror("Unable to create socket");
}
}
SocketHandler::~SocketHandler ()
{
CloseSocket();
}
int SocketHandler::InitializeAsServer ()
{
if (init_status != Uninitialized)
{
perror("Socket already initialized");
return -1;
}
struct sockaddr_in server_addr;
bzero((char *) &server_addr, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_addr.s_addr = INADDR_ANY;
server_addr.sin_port = htons(port_no);
auto bind_status = bind(socket_fd, (struct sockaddr *) &server_addr, sizeof(server_addr));
if (bind_status < 0)
{
perror("Socket binding not successful");
}
return bind_status;
}
void SocketHandler::ConnectToClient ()
{
is_client_connected = true;
if (InitializeAsServer() >= 0)
{
puts("Connecting to client");
listen(socket_fd, 1);
auto size_of_addr = sizeof(connect_to_addr);
bzero((char *) &connect_to_addr, size_of_addr);
connected_socket_fd = accept(socket_fd, (struct sockaddr *) &connect_to_addr, (socklen_t *) &size_of_addr);
close(socket_fd);
printf("Connection established. Previous socket(%d) replaced by socket(%d)\n", socket_fd, connected_socket_fd);
}
}
void SocketHandler::ConnectToServer (std::string ip_address)
{
printf("Connecting to server on port %d. Socket(%d)\n", port_no, socket_fd);
if (init_status != Uninitialized)
{
fputs("Socket already initialized\n", stderr);
}
auto client = inet_addr(ip_address.c_str());
struct sockaddr_in client_addr;
bzero((char *) &client_addr, sizeof(client_addr));
client_addr.sin_family = AF_INET;
client_addr.sin_addr.s_addr = client;
client_addr.sin_port = htons(port_no);
if (connect(socket_fd, (struct sockaddr *) &client_addr, sizeof(client_addr)) < 0)
{
perror("Cannot connect to server");
}
}
ssize_t SocketHandler::SendLarge (const void * data, std::size_t len)
{
auto dataOffset = 0;
while (len > 0)
{
auto sentSize = Send(static_cast<const char *>(data) + dataOffset, PACKET_SIZE < len ? PACKET_SIZE : len);
if (sentSize < 0)
{
perror("Message not sent by SocketHandler::SendLarge");
return -1;
}
len -= sentSize;
dataOffset += sentSize;
}
return dataOffset;
}
ssize_t SocketHandler::ReceiveLarge (void * buffer, std::size_t len)
{
auto dataOffset = 0;
while (len > 0)
{
auto receiveSize = Receive(static_cast<char *>(buffer) + dataOffset, PACKET_SIZE < len ? PACKET_SIZE : len);
if (receiveSize < 0)
{
perror("Message not received by SocketHandler::ReceiveLarge");
return -1;
}
len -= receiveSize;
dataOffset += receiveSize;
}
return dataOffset;
}
ssize_t SocketHandler::SendAndWait (const void * data, std::size_t len)
{
auto retStatus = Send(data, len);
char recAck;
Receive(&recAck);
return retStatus;
}
ssize_t SocketHandler::ReceiveAndAcknowledge (void * buffer, std::size_t len, std::function <void(void *)> PerformBeforeAcknowledging)
{
struct sockaddr_storage read_from;
socklen_t read_from_len = sizeof(read_from);
struct pollfd ds[1];
ds[0].fd = GetFileDescriptor();
ds[0].events = POLLIN;
poll(ds, 1, -1);
auto retStatus = recvfrom(GetFileDescriptor(), buffer, len, 0, (struct sockaddr *)&read_from, &read_from_len);
PerformBeforeAcknowledging(buffer);
auto recAck = 'A';
sendto(GetFileDescriptor(), &recAck, sizeof(recAck), 0, (struct sockaddr *)&read_from, read_from_len);
return retStatus;
}
This class can be used for both of the packet (UDP) and streaming (TCP) communication protocols; the user just needs to specify socket type in the constructor. Network communication usually takes place between a client and a server. The method SocketHandler::InitializeAsServer is used to set up a UDP or TCP server, but as TCP communication is connection-based, the server must also use the method SocketHandler::ConnectToClient to accept the TCP connection request from the client. This method is not useful for UDP communication. On the side of the client, UDP or TCP client sockets must make a call to SocketHandler::ConnectToServer during initialization. Sending and receiving of data is accomplished by the rest of the methods whose names are indicative of purpose. Usage will become more evident as we invoke them in the main driver.
Let us not forget—our goal is to stream video (send frames) over a network. A known thing about video frames by digital artists is that they are large. A system imposed limitation for the capacity of send() and recv() being in place, we must devise means for sending and receiving very large byte sizes. The methods SocketHandler::SendLarge and SocketHandler::ReceiveLarge serve that purpose. These methods work by breaking a large packet into smaller chunks of PACKET_SIZE=21845. This is the optimum size for the setup that I have and was experimentally determined by observing in Wireshark that, for this size, there is a one-to-one correspondence between the packet sent and acknowledgement received in a TCP communication. There is a common misconception that, as in UDP, TCP requires to have a “receive” for each “send” in a one-to-one correspondence. Actually, TCP is a streaming protocol, and, as such, the receiver can collect its data at whatever convenient time. We can have two sends and one receive, for example. The methods SocketHandler::SendLarge and SocketHandler::ReceiveLarge reflect this fact.
One of the ways we can coordinate between different processes is by message passing and waiting for a response or acknowledgement. The method SocketHandler::SendAndWait will send a message to the other end and immediately go into a state of wait. It will come out after receiving an acknowledgement. The method SocketHandler::ReceiveAndAcknowledge, on the other end, would have received the message that was sent and would acknowledge to the sender of the message. But, not before it performs certain actions! To perform the actions, a function or any other callable object can be passed to the method SocketHandler::ReceiveAndAcknowledge by its parameter PerformBeforeAcknowledging.
The other way that we can send a clear message to a process is by the use of signals. The operating system is able to deliver many signals to the processes. As an example, pressing Cntrl+C allows the operating system to deliver a blow of SIGINT (interrupt) to the process in the active terminal, SIGALRM is raised by a system call alarm(). But, the signals that are of more interest to the current context are SIGPIPE and SIGIO. SIGPIPE is produced when a TCP connection breaks and attempts to communicate over that connection continue, and SIGIO is the signal that the operating system generates to indicate the readiness of an IO device (by means of its file descriptor) for transmitting or receiving data.. Firstly, though, let us learn how to register signal handlers for the known signals.
#ifndef SIGNALHANDLING_H
#define SIGNALHANDLING_H
#include <csignal>
#include <stdio.h>
#include <sys/fcntl.h>
#include <unistd.h>
struct SignalHandling
{
void operator() (int signal, void performSigAction (int s), int fd = -1)
{
struct sigaction sigParam;
if (sigfillset(&sigParam.sa_mask) < 0)
{
perror("SignalHandling::PrepareSIGIO cannot block other signals");
}
sigParam.sa_flags = 0;
sigParam.sa_handler = performSigAction;
if (sigaction(signal, &sigParam, NULL) < 0)
{
perror("SignalHandling::PrepareSIGIO unable to change default behavior after signal delivery");
}
if (signal == SIGIO)
{
const auto CURRENT_PROCESS = getpid();
if (fcntl(fd, F_SETOWN, CURRENT_PROCESS) < 0)
{
perror("SignalHandling::PrepareSIGIO unable to set ownership of descriptor's SIGIO to current process");
}
if (fcntl(fd, F_SETFL, O_ASYNC) < 0)
{
perror("SignalHandling::PrepareSIGIO unable to set file descriptor for asynchronous delivery of signals");
}
}
}
};
SignalHandling RegisterHandlerForSignal;
#endif
The type SignalHandling overloads operator(): an object of this type is said to be a callable object. The object RegisterHandlerForSignal is declared of type SignalHandling. This object uses the system function sigaction() to register a handler for a particular signal. If no user-defined handler was provided, the signal would either be completely ignored or the default action would take effect. Another system call sigfillset() is needed here to mask off or block all other signals while the current signal handler is running. We can, also, see that an invocation of RegisterHandlerForSignal() will have a special dealing with SIGIO. For SIGIO, we are making special calls to fcntl() to set ownership of the signal to the current process and make IO operations of the file descriptor asynchronous. Similar to a regular function call, in a synchronous operation, the caller waits for the operation to return. This is in contrast to asynchronous operation where the caller continues to run until a time when the callee is ready to return. Setting ownership of the signal to the current process lets kernel know to deliver the signal to the current process, and, since we are messing with the normal flow from SIGIO to send() or recv(), asynchronous operation is what saves us from blocking on the IO operation.
A piece of advice—I cannot overemphasize the importance of descriptive error messages to save some of the time spent debugging.
Structure and Organization of the Project
You cannot call a C++ project complete without reckoning with its compilation and linking. Linux distributions come with a program called make that is helpful in setting target files (targets of compilation such as executables and object files) and dependencies. The program make identifies changes in the dependencies and reruns target builds accordingly. A makefile can be created to save off commands for reuse.
# filename - makefile
all : SocketHandler.o videoSendBin videoReceiveBin
SocketHandler.o : SocketHandler.h SocketHandler.cpp
g++ -Wall -g -std=c++14 -c SocketHandler.cpp
export LIBRARY_PATH=../../opencv-3.1.0/build/lib/:../../opencv-3.1.0/3rdparty/ippicv/unpack/ippicv_lnx/lib/intel64/
export PKG_CONFIG_PATH=../../opencv-3.1.0/build/unix-install/
INC=-I../../opencv-3.1.0/include \
-I../../opencv-3.1.0/modules/core/include \
-I../../opencv-3.1.0/modules/imgproc/include \
-I../../opencv-3.1.0/modules/photo/include \
-I../../opencv-3.1.0/modules/video/include \
-I../../opencv-3.1.0/modules/features2d/include \
-I../../opencv-3.1.0/modules/flann/include \
-I../../opencv-3.1.0/modules/objdetect/include \
-I../../opencv-3.1.0/modules/calib3d/include \
-I../../opencv-3.1.0/modules/imgcodecs/include \
-I../../opencv-3.1.0/modules/videoio/include \
-I../../opencv-3.1.0/modules/highgui/include \
-I../../opencv-3.1.0/modules/ml/include
videoSendBin : SocketHandler.o videoSender.cpp SignalHandling.h SharedStructs.h process_metrics.h
g++ -Wall -g -std=c++14 -o videoSendBin videoSender.cpp SocketHandler.o `pkg-config --libs opencv` $(INC)
videoReceiveBin : SocketHandler.o videoReceiver.cpp SignalHandling.h SharedStructs.h process_metrics.h
g++ -Wall -g -std=c++14 -pthread -o videoReceiveBin videoReceiver.cpp SocketHandler.o `pkg-config --libs opencv` $(INC)
clean :
-rm videoSendBin videoReceiveBin SocketHandler.o
There are five targets in this makefile: clean, all, videoSendBin, videoReceiveBin, and SocketHandler.o. The rule for target clean, simply, deletes the generated binaries, and the target has no dependencies. This target is run when you want to get a fresh build of all the binaries. In a terminal prompt that has access to the makefile, run
> make clean
Target all builds all of the binaries. It has, as dependency, videoSendBin, videoReceiveBin, and SocketHandler.o. In command-line, enter
> make
Targets videoSendBin, videoReceiveBin, and SocketHandler.o are similar in the making. They all have source files as dependents and summon g++ (the compiler and linker program) with the compiler and linker options. But, videoSendBin and videoReceiveBin are final linked products and are different from SocketHandler.o (object file that is to be linked into the executables).
The executables videoSendBin and videoReceiveBin are linked against OpenCV. But, OpenCV was installed in a non-standard prefix; the command-line environment must be modified using the environment variables. The modified environment variable LIBRARY_PATH makes the OpenCV shared libraries available for our builds, and PKG_CONFIG_PATH brings forward various information about the package such as finding where the shared libraries are by invoking `pkg-config --libs opencv`. The other options used with g++ are -I (include header files), -Wall (do not ignore any build warnings), -g (make binary GDB debuggable), -std=c++14 (use C++14 standard) -pthread (make multi-threading available), and -o (name the output binary).
The picture demonstrates how the project is structured. You have already encountered many of the files, but you can see a couple of new ones. The video files bird.avi and nasa_arctic_sea_ice_sets_new_record_winter_low.mp4 are the files that I used for testing and experimenting. As a consequence of the choice to install OpenCV in a non-standard prefix, we must modify yet another environment variable, LD_LIBRARY_PATH, to facilitate runtime linking against OpenCV shared libraries. The GDB files are there to start the executables in GDB.
export LD_LIBRARY_PATH=../../opencv-3.1.0/build/lib/
./videoSendBin $1
export LD_LIBRARY_PATH=../../opencv-3.1.0/build/lib/
gdb ./videoSendBin $1
export LD_LIBRARY_PATH=../../opencv-3.1.0/build/lib/
./videoReceiveBin
export LD_LIBRARY_PATH=../../opencv-3.1.0/build/lib/
gdb ./videoReceiveBin
Sending and Receiving Video Frames
Data communication is made possible by a protocol; the sender and receiver must agree on a protocol. The protocol determines how the data should be formatted to be correctly interpreted by the receiver. After the sending process sends the frames, on the receiving end the frames sent via the TCP connection need to be received, put back together, and displayed. But, to reassemble the block of memory that is supposed to be the matrix representation of the video frame, we need meta-data such as row and column size. We are going to send such data as the dimensions of the matrix representation over UDP. The sending and receiving processes share SharedStructs to pass frame metadata.
#ifndef SHAREDSTRUCTS_H
#define SHAREDSTRUCTS_H
struct MatMetaData
{
int rows;
int cols;
int type;
};
#endif
The TCP or UDP receiving functions can be blocking meaning that they will wait forever in anticipation of the arrival of the next packet. Therefore, the UDP and TCP servers cannot run single-threaded; we have to devise a multi-threaded solution. One of them blocking should not prevent the other one from executing. Here is the Receiver.
#include "SharedStructs.h"
#include "SignalHandling.h"
#include "SocketHandler.h"
#include <chrono>
#include <condition_variable>
#include <functional>
#include <future>
#include <iostream>
#include <memory>
#include <mutex>
#include <thread>
#include <opencv2/opencv.hpp>
#include "process_metrics.h"
std::mutex tcpReadyLocker;
std::condition_variable tcpReadyEvent;
auto tcpReadyNonSpurious = false;
std::mutex udpReadyLocker;
std::condition_variable udpReadyEvent;
auto udpReadyNonSpurious = false;
std::promise<MatMetaData> * frameMeta = nullptr;
std::function<void ()> ConnectionReset;
void ReceiverMainEntry()
{
std::cout << "Video receiver ..." << std::endl;
while (true)
{
SocketHandler tcpSocket(SOCK_STREAM, PORT_ONE);
auto AsyncConnect = [&tcpSocket]{ tcpSocket.ConnectToClient(); return &tcpSocket; };
auto futureConnection = std::async(std::launch::async, AsyncConnect);
do
{
std::this_thread::sleep_for(std::chrono::milliseconds(200));
}while (!frameMeta);
auto frameMetaData = frameMeta->get_future().get();
frameMeta = nullptr;
auto dispFrame = cv::Mat{frameMetaData.rows, frameMetaData.cols, frameMetaData.type};
auto frameSize = dispFrame.total() * dispFrame.elemSize();
auto framePixels = std::make_unique<uchar[]>(frameSize);
dispFrame.data = framePixels.get();
{
std::lock_guard<std::mutex> lk(tcpReadyLocker);
tcpReadyNonSpurious = true;
}
tcpReadyEvent.notify_one();
futureConnection.get();
ConnectionReset = [&tcpSocket]{ tcpSocket.CloseSocket(); };
cv::namedWindow("Video Receive", cv::WINDOW_AUTOSIZE);
while (true)
{
if (tcpSocket.ReceiveLarge(dispFrame.data, frameSize) >= 0)
{
imshow("Video Receive", dispFrame);
}
else
{
std::this_thread::sleep_for(std::chrono::seconds(5));
break;
}
if (cv::waitKey(30) == 0)
{
puts("Key press");
break;
}
}
}
}
void MetaReceiverEntry()
{
SocketHandler udpSocket(SOCK_DGRAM, PORT_TWO);
udpSocket.InitializeAsServer();
auto action = [] (int)
{
static auto passOnce = true;
if (passOnce)
{
passOnce = false;
return;
}
{
std::lock_guard<std::mutex> lk(udpReadyLocker);
udpReadyNonSpurious = true;
}
udpReadyEvent.notify_one();
};
RegisterHandlerForSignal(SIGIO, action, udpSocket.GetFileDescriptor());
MatMetaData frameInfo;
while (true)
{
std::promise<MatMetaData> p;
frameMeta = &p;
auto PerformToAcknowledge = [] (void * frameData)
{
frameMeta->set_value(*static_cast<MatMetaData *>(frameData));
std::unique_lock<std::mutex> lk(tcpReadyLocker);
tcpReadyEvent.wait(lk, []
{
auto returnValue = tcpReadyNonSpurious;
tcpReadyNonSpurious = false;
return returnValue;
});
};
udpSocket.ReceiveAndAcknowledge(&frameInfo, sizeof(frameInfo), PerformToAcknowledge);
std::unique_lock<std::mutex> lk(udpReadyLocker);
udpReadyEvent.wait(lk, []
{
auto returnValue = udpReadyNonSpurious;
udpReadyNonSpurious = false;
return returnValue;
});
ConnectionReset();
}
}
int main()
{
std::thread bgthread(MetaReceiverEntry);
bgthread.detach();
ReceiverMainEntry();
return 0;
}
The function MetaReceiverEntry() is started in a new thread and runs the UDP server. Control of the thread is immediately passed to the operating system to run it in the background by the construct bgthread.detach(). The UDP server is always ready to receive metadata from the Sender; it will update the main thread with the received information and acknowledge to the Sender. It passes the metadata by delivering a std::promise to a std::future in the main thread. If you are wondering why the std::promise is declared as a pointer, it is because you cannot reset and reuse a std::promise; you always have to declare a std::promise object anew. Going back to the point of acknowledgement, before acknowledgement, we are going to make sure that the TCP connection is well established and dynamic memory is allocated in the main thread by running the lambda function PerformToAcknowledge(); here, we wait for tcpReadyEvent. After acknowledgement, the Receiver will enter into a mode of waiting for the next connection request (from another Sender). It waits for the event of the handler of SIGIO to be triggered. The lambda function action() is the SIGIO signal handler and awakens udpReadyEvent. At this time, the currently existing TCP connection will be severed by a call to ConnectionReset(), and the process will start allover again.
One small concern is cleanly exiting the application. The resources such as network sockets and heap memory must be relinquished and returned to the operating system. The Sender can be made to exit once it completes sending the video frames. But, the Receiver, since it can span multiple video sends, must coordinate with the Sender. The TCP server is started by ReceiverMainEntry() and runs in the main thread. After receiving metadata from the UDP server, an OpenCV matrix object called dispFrame is created. The pixels to be displayed are stored in dynamic memory managed by the smart pointer framePixels. The beauty of smart pointers is that you do not have to worry about deallocating the managed memory as they will automatically do it for you when they go out of scope. And, to close the TCP connection at the appropriate time, ConnectionReset() is declared as std::function, assigned a callable object, and passed to the UDP server (remember the UDP server knows when to use it). The TCP connection was , in the first place, asynchronously established as by the object AsyncConnect because we know that waiting on a client can be blocking.
Next, we see what happened to the Sender, eventually.
#include "SharedStructs.h"
#include "SignalHandling.h"
#include "SocketHandler.h"
#include <iostream>
#include <opencv2/opencv.hpp>
#include "process_metrics.h"
int SenderMainEntry (const char* videoFile)
{
std::cout << "Video sender ..." << std::endl;
cv::VideoCapture capturedVideo;
capturedVideo.open(videoFile);
if (!capturedVideo.isOpened())
{
std::cerr << "Unable to open video: " << videoFile << std::endl;
return -1;
}
cv::namedWindow("Video Send", cv::WINDOW_AUTOSIZE);
SocketHandler udpSocket(SOCK_DGRAM, PORT_TWO);
udpSocket.ConnectToServer("127.0.0.1");
SocketHandler tcpSocket(SOCK_STREAM, PORT_ONE);
auto action = [] (int) { fputs("Exiting in response to SIGPIPE\n", stderr); exit(1); };
RegisterHandlerForSignal(SIGPIPE, action);
cv::Mat dispFrame;
std::size_t frameSize;
bool passOnce = false;
while (true)
{
capturedVideo >> dispFrame;
if (!passOnce)
{
passOnce = true;
MatMetaData frameMetaData{dispFrame.rows, dispFrame.cols, dispFrame.type()};
udpSocket.SendAndWait(&frameMetaData, sizeof(frameMetaData));
tcpSocket.ConnectToServer("127.0.0.1");
frameSize = dispFrame.total() * dispFrame.elemSize();
}
if (dispFrame.empty())
{
break;
}
else
{
tcpSocket.SendLarge(dispFrame.data, frameSize);
imshow("Video Send", dispFrame);
}
if (cv::waitKey(30) == 0)
{
puts("Key press");
break;
}
}
return 0;
}
int main(int argv, char** argc)
{
return SenderMainEntry(argc[1]);
}
Finally, to see the video run in the sender and receiver, start the receiver first by entering in command-line
> ./videoReceive
Next, enter in a different terminal
> videoSend bird.avi
