ft/skyscraper8
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Added an optional feature to procecss MPE Packets multi-threaded.
All checks were successful
🚀 Pack skyscraper8 / make-zip (push) Successful in 1m0s
Details

Browse Source
This commit is contained in:
feyris-tan 2026-06-25 21:59:06 +02:00
parent e86a89846d
commit 0cddb3b085
1 changed files with 94 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

135
skyscraper8/Dvb/DataBroadcasting/MultiprotocolEncapsulationDecoder.cs
View File

@ -17,64 +17,117 @@ namespace skyscraper5.Dvb.DataBroadcasting
class MultiprotocolEncapsulationDecoder : IPsiProcessor class MultiprotocolEncapsulationDecoder : IPsiProcessor
{ {
private static readonly ILog logger = LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType.Name); private static readonly ILog logger = LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType.Name);
private const bool ENABLE_MULTITHREADING = false;
public IMultiprotocolEncapsulationEventHandler EventHandler { get; } public IMultiprotocolEncapsulationEventHandler EventHandler { get; }
public MultiprotocolEncapsulationDecoder(IMultiprotocolEncapsulationEventHandler eventHandler) public MultiprotocolEncapsulationDecoder(IMultiprotocolEncapsulationEventHandler eventHandler)
{ {
EventHandler = eventHandler; EventHandler = eventHandler;
restartRequired = true;
} }
public void GatherPsi(PsiSection section, int sourcePid) public void GatherPsi(PsiSection section, int sourcePid)
{ {
MemoryStream ms = new MemoryStream(section.GetDataCopy(), false); if (!ENABLE_MULTITHREADING)
if (ms.Length < 12)
return;
byte tableId = ms.ReadUInt8();
ushort readUInt16Be = ms.ReadUInt16BE();
bool sectionSyntaxIndicator = (readUInt16Be & 0x8000) != 0;
bool privateIndicator = (readUInt16Be & 0x4000) != 0;
int sectionLength = readUInt16Be & 0x0fff;
byte[] macAddress = new byte[6];
macAddress[5] = ms.ReadUInt8();
macAddress[4] = ms.ReadUInt8();
byte readUInt8 = ms.ReadUInt8();
int payloadScramblingControl = (readUInt8 & 0x30) >> 4;
if (payloadScramblingControl != 0)
return; //We don't know how to unscramble these.
int addressScramblingControl = (readUInt8 & 0x0c) >> 2;
bool llcSnap = (readUInt8 & 0x02) != 0;
bool currentNextIndicator = (readUInt8 & 0x01) != 0;
byte sectionNumber = ms.ReadUInt8();
byte lastSectionNumber = ms.ReadUInt8();
macAddress[3] = ms.ReadUInt8();
macAddress[2] = ms.ReadUInt8();
macAddress[1] = ms.ReadUInt8();
macAddress[0] = ms.ReadUInt8();
long length = ms.GetAvailableBytes() - 4;
if (length <= 0)
{ {
return; GatherPsiEx(section, sourcePid);
}
EnsureVirtualNetworkIdentifier(sourcePid);
byte[] payload = ms.ReadBytes(length);
if (llcSnap)
{
HandleLlcSnap(sourcePid, macAddress, payload);
} }
else else
{ {
HandleIpDatagram(sourcePid,payload); if (_actionQueue == null)
} _actionQueue = new Queue<Action>();
lock (_actionQueue)
{
_actionQueue.Enqueue(() => { GatherPsiEx(section, sourcePid); });
}
if (restartRequired)
{
restartRequired = false;
_actionQueueThread = new Thread(SeperatedThreadFunction);
_actionQueueThread.Name = String.Format("MPE Decoder Thread for PID 0x{0:X4}", sourcePid);
_actionQueueThread.Start();
}
}
} }
private void GatherPsiEx(PsiSection section, int sourcePid)
{
MemoryStream ms = new MemoryStream(section.GetDataCopy(), false);
if (ms.Length < 12)
return;
byte tableId = ms.ReadUInt8();
ushort readUInt16Be = ms.ReadUInt16BE();
bool sectionSyntaxIndicator = (readUInt16Be & 0x8000) != 0;
bool privateIndicator = (readUInt16Be & 0x4000) != 0;
int sectionLength = readUInt16Be & 0x0fff;
byte[] macAddress = new byte[6];
macAddress[5] = ms.ReadUInt8();
macAddress[4] = ms.ReadUInt8();
byte readUInt8 = ms.ReadUInt8();
int payloadScramblingControl = (readUInt8 & 0x30) >> 4;
if (payloadScramblingControl != 0)
return; //We don't know how to unscramble these.
int addressScramblingControl = (readUInt8 & 0x0c) >> 2;
bool llcSnap = (readUInt8 & 0x02) != 0;
bool currentNextIndicator = (readUInt8 & 0x01) != 0;
byte sectionNumber = ms.ReadUInt8();
byte lastSectionNumber = ms.ReadUInt8();
macAddress[3] = ms.ReadUInt8();
macAddress[2] = ms.ReadUInt8();
macAddress[1] = ms.ReadUInt8();
macAddress[0] = ms.ReadUInt8();
long length = ms.GetAvailableBytes() - 4;
if (length <= 0)
{
return;
}
EnsureVirtualNetworkIdentifier(sourcePid);
byte[] payload = ms.ReadBytes(length);
if (llcSnap)
{
HandleLlcSnap(sourcePid, macAddress, payload);
}
else
{
HandleIpDatagram(sourcePid, payload);
}
}
private Thread _actionQueueThread;
private Queue<Action> _actionQueue;
private bool restartRequired;
private void SeperatedThreadFunction()
{
while (true)
{
int remainingElements = 0;
lock (_actionQueue)
{
remainingElements = _actionQueue.Count;
}
if (remainingElements == 0)
{
break;
}
Action action = null;
lock (_actionQueue)
{
action = _actionQueue.Dequeue();
}
action();
}
restartRequired = true;
}
private bool announcedLlcTraffic; private bool announcedLlcTraffic;
private void HandleLlcSnap(int sourcePid, byte[] macAddress, byte[] payload) private void HandleLlcSnap(int sourcePid, byte[] macAddress, byte[] payload)