Как я могу использовать ConcurrentQueue для написания в файл в отдельном потоке? ⇐ C#

[Anonymous]

Мое приложение получает данные, которые я пытаюсь написать в файл в отдельном потоке (написание их в потоке, которое получило его непосредственно, вызвало проблемы с производительностью). Кажется, для этого подходит CondurentQueue . Но написание в файл вообще не происходит: < /p>
Объявления: < /p>
// Declare the queue as class members
private static ConcurrentQueue queue = new ConcurrentQueue();
private static Task consumerTask;
< /code>
Consumer: < /p>
static void Consumer()
{
// Open the binary file for writing
using (var fileStream = File.OpenWrite("received_data.bin"))
{
// Try to dequeue an item from the queue
if (queue.TryDequeue(out var dataBuffer))
{
// Write the data to the binary file
fileStream.Write(dataBuffer.Buffer, 0, dataBuffer.Length);
}
}
}
< /code>
Зачалась задача потребителя при открытии внешнего интерфейса, поэтому ожидаются данные: < /p>
consumerTask = Task.Factory.StartNew(Consumer);
< /code>
Данные Enqueue в обработчике событий, выполненном при получении данных: < /p>
// Enqueue the received data in the queue
queue.Enqueue(new DataBuffer(e.Bytes, (int)e.NumBytesAvailable));
< /code>
DataBuffer класс (передача данных в очередь и его размер):
public class DataBuffer
{
public byte[] Buffer { get; set; }
public int Length { get; set; }

public DataBuffer(byte[] buffer, int length)
{
Buffer = buffer;
Length = length;
}
}
< /code>
I add AutoResetEvent и запись происходит только один раз, для одного полученного пакета данных:
private static AutoResetEvent queueEvent = new AutoResetEvent(false);
< /code>
// Enqueue the received data in the queue
queue.Enqueue(new DataBuffer(e.Bytes, (int)e.NumBytesAvailable));
// Set the auto reset event to signal that there is data available in the queue
queueEvent.Set();
< /code>
Consumer waiting for data:
static void Consumer()
{
// Open the binary file for writing
using (var fileStream = File.OpenWrite("received_data.bin"))
{
// Wait until there is data available in the queue
queueEvent.WaitOne();

// Try to dequeue an item from the queue
if (queue.TryDequeue(out var dataBuffer))
{
// Write the data to the binary file
fileStream.Write(dataBuffer.Buffer, 0, dataBuffer.Length);
}
}
}
< /code>
I am trying to use BlockingCollection Но данные, полученные в Consumer , испорчены по сравнению с тем, что добавлено в сборку. Как будто читается другая часть буфера данных. Что -то вроде 0x112233445566778899 ... и я получаю 0x5566778899aabbccdd ...
может быть связан с тем, что полученные данные не входят в полные пакеты, так что буфер данных передается в коллекцию, не всегда начинается с примера 0x112334455567777777 ... и, кажется, не всегда начинается с кода, и я не всегда начинается с кода, и я не всегда начинается с Compectomer
private static BlockingCollection collection = new BlockingCollection();
< /code>
Addidng data to collection:
// Add the received data to the blocking collection
collection.Add(new DataBuffer(e.Bytes, (int)e.NumBytesAvailable));
< /code>
In Consumer:
while(true)
{
if (collection.TryTake(out var dataBuffer, Timeout.Infinite))
{
// Write the data to the binary file
fileStream.Write(dataBuffer.Buffer, 0, dataBuffer.Length);
}
}
< /code>
I changed it to modern class Channel, но все же получают неправильные результаты. Вывод первых 8 байтов буферов производителя/потребителей по порядку в соответствии с обработкой:
Producer: 12-AB-34-CD-56-EF-78-56
Producer: 12-AB-34-CD-56-EF-78-56
Producer: 56-EF-78-56-9A-43-BC-21
Consumer: 56-EF-78-56-9A-43-BC-21
Consumer: 56-EF-78-56-9A-43-BC-21
Consumer: 56-EF-78-56-9A-43-BC-21
Producer: 12-AB-34-CD-56-EF-78-56
Producer: 56-9A-43-BC-21-12-AB-34
Consumer: 56-9A-43-BC-21-12-AB-34
Producer: 21-12-AB-34-CD-56-EF-78
Consumer: 21-12-AB-34-CD-56-EF-78
Producer: CD-56-EF-78-56-9A-43-BC
Consumer: CD-56-EF-78-56-9A-43-BC
Producer: 21-12-AB-34-CD-56-EF-78
< /code>
Code snippets for Channel:
private static Channel channel = Channel.CreateUnbounded();
< /code>
channel.Writer.TryWrite(new DataBuffer(_ReceivedDataBuffer2, _ReceivedRawDataBufferPosition2));
< /code>
static async Task Consumer()
{
using (var fileStream = File.OpenWrite("received_data.bin"))
{
while (await channel.Reader.WaitToReadAsync())
{
// Try to read a DataBuffer instance from the channel
if (channel.Reader.TryRead(out var dataBuffer))
{
System.Diagnostics.Debug.WriteLine("Consumer: " + BitConverter.ToString(dataBuffer.Buffer.Take(8).ToArray()));
// Write the data to the binary file
fileStream.Write(dataBuffer.Buffer, 0, dataBuffer.Length);
}
}
}
}
< /code>
I went back to using BlockingCollection. Ниже я демонстрирую, что я использую только одну ветку производителя и один потребитель. Я добавил написание данных в файл, также перед передачей его в коллекцию, чтобы я мог сравнить данные до и после очереди. В этих байтах у меня есть пакеты около 1,5 кб, состоящие из данных + № последовательности + read_status .
В сохраненных файлах я вижу различия. В файле, сохраненном потребителем я вижу правильную последовательность пакетов 41, 42, 43, после чего пакет 44 больше не является цельным, и есть прыжок, пакет 56, затем есть еще один пакет 57, 58, 59, 60, после чего возвращается в ранее пропущенные пакеты, но не все, например, 48, 49, 50, 50, 80. 59, 60 и пакеты от 45 до 47 не присутствуют. < /P>
Этот тип перегруппировки появляется в этом файле без перерыва после выхода из очереди. Данные в файле перед очередью находятся в правильном порядке и непрерывности. < /P>
using System;
using System.IO;
using System.Management;
using System.Collections.Concurrent;
using System.Threading.Tasks;

namespace test_app_1
{
public class USB
{
// -----------------------------------------------------------------
// Type of commands handled by FPGA constants:
const byte CMD_DDC_REGISTER = 0x30;

// -----------------------------------------------------------------
// Type of application operations constants:
const int MODE_NOP = 0;
const int MODE_DDC_RAW_SAVE_TO_FILE = 16;

// default operation mode:
private int _OperationMode = MODE_NOP;

// -----------------------------------------------------------------
// Events declarations:
public event EventHandler OnFtdiConnected;
public event EventHandler OnFtdiDisconnected;
public event EventHandler OnFtdiOpenedAndConfigured;

public FtdiFifo FtdiFifo;
private readonly string _AllowedSerialNumber = "FT786P7IA";
public bool FtdiOpened = false;

private byte[] _CommandBytes = new byte[10];

private byte[] _ReceivedRawDataBuffer = new byte[150000];
private uint _ReceivedNumberOfBytes = 0;

static private FileStream _FileStream;
static private FileStream _FileStreamBeforeQueue;

private static Task consumerTask;
private static BlockingCollection collection = new BlockingCollection();
private DataBuffer producerDataBuffer;

// -----------------------------------------------------------------
// USB class constructor:
// * configures automatic detection of events related to connected USB devices
// -----------------------------------------------------------------
public USB()
{
// ------------ autodetect USB hardware ------------------------
// Set up the query to listen for device arrival and removal events
WqlEventQuery query = new WqlEventQuery("SELECT * FROM __InstanceOperationEvent WITHIN 1 WHERE TargetInstance ISA 'Win32_USBHub'");

// Set up the event watcher
ManagementEventWatcher watcher = new ManagementEventWatcher(query);

// Add event handlers to be notified when a device is connected or removed
watcher.EventArrived += new EventArrivedEventHandler(Handler_DeviceChangedEvent);

// Start listening for events
watcher.Start();
}

// -----------------------------------------------------------------
// EVENT HANDLER: for USB FTDI device connection events
// -----------------------------------------------------------------
private void Handler_DeviceChangedEvent(object sender, EventArrivedEventArgs e)
{
using (var moBase = (ManagementBaseObject)e.NewEvent.Properties["TargetInstance"].Value)
{
string devicePNPId = moBase?.Properties["PNPDeviceID"]?.Value.ToString();

switch (e.NewEvent.ClassPath.ClassName)
{
case "__InstanceDeletionEvent":
//BeginInvoke(new Action(() => UpdateUI(eventMessage)));
if (devicePNPId == @"USB\VID_0403&PID_6010\" + _AllowedSerialNumber.Substring(0, _AllowedSerialNumber.Length - 1)) //remove last character (A - channel indicator) from the serial number
{
FtdiOpened = false;
// Raise the event
OnFtdiDisconnected?.Invoke(this, EventArgs.Empty);
}
break;
case "__InstanceCreationEvent":
//BeginInvoke(new Action(() => UpdateUI(eventMessage)));
if(devicePNPId == @"USB\VID_0403&PID_6010\" + _AllowedSerialNumber.Substring(0, _AllowedSerialNumber.Length - 1)) //remove last character (A - channel indicator) from the serial number
{
// Raise the event
OnFtdiConnected?.Invoke(this, new CustomEventArgs(_AllowedSerialNumber.Substring(0, _AllowedSerialNumber.Length - 1)));
}
break;
case "__InstanceModificationEvent":
default:
break;
}
}
}

// -----------------------------------------------------------------
// EVENT HANDLER: Handles event triggered when FTDI is configured
// for synchronous 245 mode; generates event for detector class
// -----------------------------------------------------------------
private void Handler_OnFtdiOpenedAndSetToFifoMode(object sender, CustomEventArgs e)
{
OnFtdiOpenedAndConfigured?.Invoke(this, new CustomEventArgs(e.Value));
}

// -----------------------------------------------------------------
// Open and configure FTDI device
// -----------------------------------------------------------------
public void OpenAndConfigureFTDI()
{
// FTDI object
FtdiFifo = new FtdiFifo();

// FTDI device detection
FtdiFifo.OnFtdiOpenedAndSetToFifoMode += Handler_OnFtdiOpenedAndSetToFifoMode;
FtdiFifo.IdentifyDevice(); // FTDI device detection
if (FtdiFifo.IsDeviceAllowed(_AllowedSerialNumber)) // Is the connected device allowed? (_AllowedSerialNumber)
{
// FTDI chip configuration to FIFO mode
FtdiFifo.OpenInFifoMode(_AllowedSerialNumber);

FtdiOpened = FtdiFifo.Opened;

// Subscription to Events:
FtdiFifo.OnFtdiBytesReceived += Handler_OnFtdiBytesReceived;

// Start the consumer task
consumerTask = Task.Run(Consumer);
}
}

// -----------------------------------------------------------------
// EVENT HANDLER: for FTDI USB data reception
// -----------------------------------------------------------------
void Handler_OnFtdiBytesReceived(object sender, FtdiBytesReceivedEventArgs e)
{
switch (_OperationMode)
{
case MODE_DDC_RAW_SAVE_TO_FILE:

_ReceivedNumberOfBytes = e.NumBytesAvailable;
Array.Copy(e.Bytes, 0, _ReceivedRawDataBuffer, 0, _ReceivedNumberOfBytes);

// Add the received data to the blocking collection
producerDataBuffer = new DataBuffer(_ReceivedRawDataBuffer, (int)_ReceivedNumberOfBytes);
collection.Add(producerDataBuffer);

// TEST PURPOSE ONLY - writing to separate file the same data which is enqueued in collection
// Data "before queue" and "after queue" can be compared later
_FileStreamBeforeQueue.Write(_ReceivedRawDataBuffer, 0, (int)_ReceivedNumberOfBytes);

break;

default:
break;
}
}
// -----------------------------------------------------------------
// A method that fetches data from a queue and writes it to a file.
// It works in a dedicated Task.
// -----------------------------------------------------------------
static void Consumer()
{
while (true)
{
foreach (DataBuffer dataBuffer in collection.GetConsumingEnumerable())
{
_FileStream.Write(dataBuffer.Buffer, 0, dataBuffer.Length);
}

}
}

// -----------------------------------------------------------------
// A method that sends a measurement start command to the device.
// Used for button click handling.
// -----------------------------------------------------------------
public void DdcEnableConv()
{
_OperationMode = MODE_DDC_RAW_SAVE_TO_FILE;

String FileName = "ddc_readout_" + DateTime.Now.ToString("yyyy_MM_dd_HH_mm_ss") + ".bin";
String FileName2 = "ddc_readout_before_queue_" + DateTime.Now.ToString("yyyy_MM_dd_HH_mm_ss") + ".bin";
_FileStream = new FileStream(FileName, FileMode.Create, FileAccess.Write);
_FileStreamBeforeQueue = new FileStream(FileName2, FileMode.Create, FileAccess.Write);

Array.Clear(_CommandBytes, 0, _CommandBytes.Length); // zeroing command array
_CommandBytes[0] = CMD_DDC_REGISTER; // setting command byte (first byte)
_CommandBytes[1] = 0x07; // setting command byte (first byte)

_CommandBytes[9] = 0x01;

FtdiFifo.SendDataIntoFifo2(_CommandBytes); // sending data through FTDI
}

// -----------------------------------------------------------------
// A method that sends a measurement stop command to the device.
// Used for button click handling.
// -----------------------------------------------------------------
public void DdcDisableConv()
{
_OperationMode = MODE_DDC_RAW_SAVE_TO_FILE;

Array.Clear(_CommandBytes, 0, _CommandBytes.Length); // zeroing command array
_CommandBytes[0] = CMD_DDC_REGISTER; // setting command byte (first byte)
_CommandBytes[1] = 0x07; // setting command byte (first byte)

_CommandBytes[9] = 0x00;

FtdiFifo.SendDataIntoFifo2(_CommandBytes); // sending data through FTDI
}

}
}

// -----------------------------------------------------------------
// A class that defines an object used to transfer measurement data
// received from the FTDI (producer) to a Task that writes the data
// to a file (consumer).
//
// Allows two arguments to be sent: the data buffer and the length
// of valid data in the buffer.
// -----------------------------------------------------------------
public class DataBuffer
{
public byte[] Buffer { get; set; }
public int Length { get; set; }

public DataBuffer(byte[] buffer, int length)
{
Buffer = buffer;
Length = length;
}
}


Подробнее здесь: https://stackoverflow.com/questions/749 ... ate-thread