307 lines
10 KiB
C#
307 lines
10 KiB
C#
using System;
|
||
using System.Collections.Generic;
|
||
using System.Linq;
|
||
using DotNetty.Buffers;
|
||
using fec;
|
||
|
||
namespace fec.fec
|
||
{
|
||
public class FecDecode
|
||
{
|
||
// queue size limit
|
||
private readonly int rxlimit;
|
||
private readonly int dataShards;
|
||
|
||
private int parityShards;
|
||
|
||
/** dataShards+parityShards **/
|
||
private readonly int shardSize;
|
||
|
||
// ordered receive queue
|
||
private readonly List<FecPacket> rx;
|
||
|
||
private readonly IByteBuffer[] decodeCache;
|
||
|
||
/**标记是否已经缓存了**/
|
||
private readonly bool[] flagCache;
|
||
|
||
private readonly ReedSolomon codec;
|
||
|
||
private readonly IByteBuffer zeros;
|
||
|
||
|
||
public FecDecode(int rxlimit, ReedSolomon codec, int mtu)
|
||
{
|
||
this.rxlimit = rxlimit;
|
||
this.dataShards = codec.getDataShardCount();
|
||
this.parityShards = codec.getParityShardCount();
|
||
this.shardSize = dataShards + parityShards;
|
||
|
||
if (dataShards <= 0 || parityShards <= 0)
|
||
{
|
||
throw new Exception("dataShards and parityShards can not less than 0");
|
||
}
|
||
|
||
if (rxlimit < dataShards + parityShards)
|
||
{
|
||
throw new Exception("");
|
||
}
|
||
|
||
this.codec = codec;
|
||
this.decodeCache = new IByteBuffer[this.shardSize];
|
||
this.flagCache = new bool[this.shardSize];
|
||
this.rx = new List<FecPacket>(rxlimit);
|
||
|
||
zeros = PooledByteBufferAllocator.Default.DirectBuffer(mtu);
|
||
zeros.WriteBytes(new byte[mtu]);
|
||
}
|
||
|
||
|
||
public List<IByteBuffer> decode(FecPacket pkt)
|
||
{
|
||
if (pkt.Flag == Fec.typeParity)
|
||
{
|
||
Snmp.snmp.FECParityShards++;
|
||
}
|
||
else
|
||
{
|
||
Snmp.snmp.FECDataShards++;
|
||
}
|
||
|
||
int n = rx.Count - 1;
|
||
int insertIdx = 0;
|
||
for (int i = n; i >= 0; i--)
|
||
{
|
||
//去重
|
||
if (pkt.Seqid == rx[i].Seqid)
|
||
{
|
||
Snmp.snmp.FECRepeatDataShards++;
|
||
pkt.release();
|
||
return null;
|
||
}
|
||
|
||
if (pkt.Seqid > rx[i].Seqid)
|
||
{
|
||
// insertion
|
||
insertIdx = i + 1;
|
||
break;
|
||
}
|
||
}
|
||
|
||
//插入 rx中
|
||
if (insertIdx == n + 1)
|
||
{
|
||
this.rx.Add(pkt);
|
||
}
|
||
else
|
||
{
|
||
rx.Insert(insertIdx, pkt);
|
||
}
|
||
|
||
//所有消息列表中的第一个包
|
||
// shard range for current packet
|
||
long shardBegin = pkt.Seqid - pkt.Seqid % shardSize;
|
||
long shardEnd = shardBegin + shardSize - 1;
|
||
|
||
//rx数组中的第一个包
|
||
// max search range in ordered queue for current shard
|
||
int searchBegin = (int) (insertIdx - pkt.Seqid % shardSize);
|
||
if (searchBegin < 0)
|
||
{
|
||
searchBegin = 0;
|
||
}
|
||
|
||
int searchEnd = searchBegin + shardSize - 1;
|
||
if (searchEnd >= rx.Count)
|
||
{
|
||
searchEnd = rx.Count - 1;
|
||
}
|
||
|
||
List<IByteBuffer> result = null;
|
||
if (searchEnd - searchBegin + 1 >= dataShards)
|
||
{
|
||
//当前包组的已收到的包数量
|
||
int numshard = 0;
|
||
//当前包组中属于数据包的数量
|
||
int numDataShard = 0;
|
||
//搜到第一个包在搜索行中的位置
|
||
int first = 0;
|
||
//收到的最大包的字节长度
|
||
int maxlen = 0;
|
||
|
||
// zero cache
|
||
IByteBuffer[] shards = decodeCache;
|
||
bool[] shardsflag = flagCache;
|
||
for (int i = 0; i < shards.Length; i++)
|
||
{
|
||
shards[i] = null;
|
||
shardsflag[i] = false;
|
||
}
|
||
|
||
for (int i = searchBegin; i <= searchEnd; i++)
|
||
{
|
||
FecPacket fecPacket = rx[i];
|
||
long seqid = fecPacket.Seqid;
|
||
if (seqid > shardEnd)
|
||
break;
|
||
if (seqid < shardBegin)
|
||
continue;
|
||
shards[(int) (seqid % shardSize)] = fecPacket.Data;
|
||
shardsflag[(int) (seqid % shardSize)] = true;
|
||
numshard++;
|
||
if (fecPacket.Flag == Fec.typeData)
|
||
{
|
||
numDataShard++;
|
||
}
|
||
|
||
if (numshard == 1)
|
||
{
|
||
first = i;
|
||
}
|
||
|
||
if (fecPacket.Data.ReadableBytes> maxlen)
|
||
{
|
||
maxlen = fecPacket.Data.ReadableBytes;
|
||
}
|
||
}
|
||
|
||
if (numDataShard == dataShards)
|
||
{
|
||
freeRange(first, numshard, rx);
|
||
}
|
||
else if (numshard >= dataShards)
|
||
{
|
||
for (int i = 0; i < shards.Length; i++)
|
||
{
|
||
IByteBuffer shard = shards[i];
|
||
//如果数据不存在 用0填充起来
|
||
if (shard == null)
|
||
{
|
||
shards[i] = zeros.Copy(0, maxlen);
|
||
shards[i].SetWriterIndex(maxlen);
|
||
continue;
|
||
}
|
||
|
||
int left = maxlen - shard.ReadableBytes;
|
||
if (left > 0)
|
||
{
|
||
shard.WriteBytes(this.zeros, left);
|
||
zeros.ResetReaderIndex();
|
||
// zeros.resetReaderIndex();
|
||
}
|
||
}
|
||
|
||
codec.decodeMissing(shards, shardsflag, 0, maxlen);
|
||
result = new List<IByteBuffer>(this.dataShards);
|
||
for (int i = 0; i < shardSize; i++)
|
||
{
|
||
if (shardsflag[i])
|
||
{
|
||
continue;
|
||
}
|
||
|
||
IByteBuffer byteBufs = shards[i];
|
||
//释放构建的parityShards内存
|
||
if (i >= dataShards)
|
||
{
|
||
byteBufs.Release();
|
||
continue;
|
||
}
|
||
|
||
int packageSize = byteBufs.ReadShort();
|
||
if (byteBufs.ReadableBytes < packageSize)
|
||
{
|
||
//// System.out.println("bytebuf长度: " + byteBufs.writerIndex() + " 读出长度" + packageSize);
|
||
// byte[] bytes = new byte[byteBufs.writerIndex()];
|
||
// byteBufs.getBytes(0, bytes);
|
||
// for (byte aByte :
|
||
// bytes) {
|
||
// System.out.print("[" + aByte + "] ");
|
||
// }
|
||
Snmp.snmp.FECErrs++;
|
||
}
|
||
else
|
||
{
|
||
Snmp.snmp.FECRecovered++;
|
||
}
|
||
|
||
//去除fec头标记的消息体长度2字段
|
||
byteBufs = byteBufs.Slice(Fec.fecDataSize, packageSize);
|
||
//int packageSize =byteBufs.readUnsignedShort();
|
||
//byteBufs = byteBufs.slice(0,packageSize);
|
||
result.Add(byteBufs);
|
||
Snmp.snmp.FECRecovered++;
|
||
//int packageSize =byteBufs.getUnsignedShort(0);
|
||
////判断长度
|
||
//if(byteBufs.writerIndex()-Fec.fecHeaderSizePlus2>=packageSize&&packageSize>0)
|
||
//{
|
||
// byteBufs = byteBufs.slice(Fec.fecHeaderSizePlus2,packageSize);
|
||
// result.add(byteBufs);
|
||
// Snmp.snmp.FECRecovered.incrementAndGet();
|
||
//}else{
|
||
// System.out.println("bytebuf长度: "+byteBufs.writerIndex()+" 读出长度"+packageSize);
|
||
// byte[] bytes = new byte[byteBufs.writerIndex()];
|
||
// byteBufs.getBytes(0,bytes);
|
||
// for (byte aByte : bytes) {
|
||
// System.out.print("["+aByte+"] ");
|
||
// }
|
||
// Snmp.snmp.FECErrs.incrementAndGet();
|
||
//}
|
||
}
|
||
|
||
freeRange(first, numshard, rx);
|
||
}
|
||
}
|
||
|
||
if (rx.Count> rxlimit)
|
||
{
|
||
if (rx[0].Flag == Fec.typeData)
|
||
{
|
||
Snmp.snmp.FECShortShards++;
|
||
}
|
||
|
||
freeRange(0, 1, rx);
|
||
}
|
||
|
||
return result;
|
||
}
|
||
|
||
|
||
public void release(){
|
||
this.parityShards=0;
|
||
foreach (var fecPacket in this.rx)
|
||
{
|
||
fecPacket?.release();
|
||
}
|
||
this.zeros.Release();
|
||
}
|
||
/**
|
||
* 1，回收first后n个bytebuf
|
||
* 2，将q的first到first+n之间的数据移除掉
|
||
* 3，将尾部的n个数据的data清空
|
||
* 4，返回开头到尾部n个数组的对象
|
||
*
|
||
* @param first
|
||
* @param n
|
||
* @param q
|
||
*/
|
||
private static void freeRange(int first,int n,List<FecPacket> q){
|
||
for (int i = first; i < first + n; i++) {
|
||
q[i].release();
|
||
}
|
||
//copy(q[first:], q[first+n:])
|
||
for (int i = first; i < q.Count; i++) {
|
||
int index = i+n;
|
||
if(index==q.Count)
|
||
break;
|
||
q[i]=q[index];
|
||
}
|
||
//for (int i = 0; i < n; i++) {
|
||
// q.get(q.size()-1-i).setData(null);
|
||
//}
|
||
for (int i = 0; i < n; i++) {
|
||
q.RemoveAt(q.Count-1);
|
||
}
|
||
}
|
||
}
|
||
} |