multithreading - threads - Sincronización de carga de archivos multiproceso

python3 multithreading

Question

Actualmente estoy trabajando en una aplicación cliente / servidor Delphi XE3 para transferir archivos (con los componentes de Indy FTP). La parte del cliente supervisa una carpeta, obtiene una lista de los archivos que contiene, los sube al servidor y elimina los originales. La carga se realiza mediante un hilo separado, que procesa los archivos uno por uno. Los archivos pueden tener un rango de 0 a algunos miles y sus tamaños también varían mucho.

Es una aplicación Firemonkey compilada tanto para OSX como para Windows, así que tuve que usar TThread en lugar de OmniThreadLibrary, que prefería. Mi cliente informa que la aplicación se congela al azar. No pude duplicarlo, pero como no tengo mucha experiencia con TThread, podría haber puesto la condición de punto muerto en alguna parte. Leí bastantes ejemplos, pero aún no estoy seguro sobre algunos de los detalles multithread.

La estructura de la aplicación es simple:
Un temporizador en el hilo principal comprueba la carpeta y obtiene información sobre cada archivo en un registro, que va a un TList genérico. Esta lista contiene información sobre los nombres de los archivos, el tamaño, el progreso, si el archivo está completamente cargado o si se debe volver a intentar. Todo lo que se muestra en una grilla con barras de progreso, etc. A esta lista solo se accede por el hilo principal. Después de eso, los elementos de la lista se envían al hilo llamando al método AddFile (código a continuación). El hilo almacena todos los archivos en una cola segura para subprocesos como esta http://delphihaven.wordpress.com/2011/05/06/using-tmonitor-2/
Cuando se carga el archivo, el subproceso de carga notifica al hilo principal con una llamada a Sincronizar.
El hilo principal llama periódicamente al método Uploader.GetProgress para verificar el progreso del archivo actual y mostrarlo. Esta función no es realmente segura para subprocesos, pero ¿podría provocar un punto muerto o solo devolver datos incorrectos?

¿Cuál sería una forma segura y eficiente de hacer la verificación de progreso?

Entonces, ¿este enfoque está bien o me he perdido algo? ¿Cómo harías esto?
Por ejemplo, pensé en hacer un nuevo hilo solo para leer el contenido de la carpeta. Esto significa que el TList que uso debe estar seguro para subprocesos, pero se debe acceder a él todo el tiempo para actualizar la información mostrada en la grilla de la GUI. ¿No sería toda la sincronización solo ralentizaría la GUI?

He publicado el código simplificado a continuación en caso de que alguien quiera verlo. Si no, me gustaría escuchar algunas opiniones sobre lo que debería usar en general. Los objetivos principales son trabajar tanto en OSX como en Windows; para poder mostrar información sobre todos los archivos y el progreso del actual; y ser receptivo independientemente del número y tamaño de los archivos.

Ese es el código del subproceso de carga. He eliminado algunos de ellos para facilitar la lectura:

type
  TFileStatus = (fsToBeQueued, fsUploaded, fsQueued);
  TFileInfo = record
    ID: Integer;
    Path: String;
    Size: Int64;
    UploadedSize: Int64;
    Status: TFileStatus;
  end;

  TUploader = class(TThread)
  private
    FTP: TIdFTP;
    fQueue: TThreadedQueue<TFileInfo>;
    fCurrentFile: TFileInfo;
    FUploading: Boolean;
    procedure ConnectFTP;
    function UploadFile(aFileInfo: TFileInfo): String;
    procedure OnFTPWork(ASender: TObject; AWorkMode: TWorkMode; AWorkCount: Int64);
    procedure SignalComplete;
    procedure SignalError(aError: String);
  protected
    procedure Execute; override;
  public
    property Uploading: Boolean read FUploading;
    constructor Create;
    destructor Destroy; override;
    procedure Terminate;
    procedure AddFile(const aFileInfo: TFileInfo);
    function GetProgress: TFileInfo;
  end;

procedure TUploader.AddFile(const aFileInfo: TFileInfo);
begin
  fQueue.Enqueue(aFileInfo);
end;

procedure TUploader.ConnectFTP;
begin
  ...
    FTP.Connect;
end;

constructor TUploader.Create;
begin
  inherited Create(false);
  FreeOnTerminate := false;
  fQueue := TThreadedQueue<TFileInfo>.Create;
  // Create the TIdFTP and set ports and other params
  ...
end;

destructor TUploader.Destroy;
begin
  fQueue.Close;
  fQueue.Free;
  FTP.Free;
  inherited;
end;

// Process the whole queue and inform the main thread of the progress
procedure TUploader.Execute;
var
  Temp: TFileInfo;
begin
  try
    ConnectFTP;
  except
    on E: Exception do
      SignalError(E.Message);
  end;

  // Use Peek instead of Dequeue, because the item should not be removed from the queue if it fails
  while fQueue.Peek(fCurrentFile) = wrSignaled do
    try
      if UploadFile(fCurrentFile) = '''' then
      begin
        fQueue.Dequeue(Temp);  // Delete the item from the queue if succesful
        SignalComplete;
      end;
    except
      on E: Exception do
        SignalError(E.Message);
    end;
end;

// Return the current file''s info to the main thread. Used to update the progress indicators
function TUploader.GetProgress: TFileInfo;
begin
  Result := fCurrentFile;
end;

// Update the uploaded size for the current file. This information is retrieved by a timer from the main thread to update the progress bar
procedure TUploader.OnFTPWork(ASender: TObject; AWorkMode: TWorkMode; AWorkCount: Int64);
begin
  fCurrentFile.UploadedSize := AWorkCount;
end;

procedure TUploader.SignalComplete;
begin
  Synchronize(
    procedure
    begin
      frmClientMain.OnCompleteFile(fCurrentFile);
    end);
end;

procedure TUploader.SignalError(aError: String);
begin
  try
    FTP.Disconnect;
  except
  end;
  if fQueue.Closed then
    Exit;

  Synchronize(
    procedure
    begin
      frmClientMain.OnUploadError(aError);
    end);
end;

// Clear the queue and terminate the thread
procedure TUploader.Terminate;
begin
  fQueue.Close;
  inherited;
end;

function TUploader.UploadFile(aFileInfo: TFileInfo): String;
begin
  Result := ''Error'';
  try
    if not FTP.Connected then
      ConnectFTP;
    FUploading := true;
    FTP.Put(aFileInfo.Path, ExtractFileName(aFileInfo.Path));     
    Result := '''';
  finally
    FUploading := false;
  end;
end;

Y partes del hilo principal que interactúan con el cargador:

......
// Main form
    fUniqueID: Integer;  // This is a unique number given to each file, because there might be several with the same names(after one is uploaded and deleted)
    fUploader: TUploader;         // The uploader thread
    fFiles: TList<TFileInfo>;
    fCurrentFileName: String;     // Used to display the progress
    function IndexOfFile(aID: Integer): Integer;    //Return the index of the record inside the fFiles given the file ID
  public
    procedure OnCompleteFile(aFileInfo: TFileInfo);
    procedure OnUploadError(aError: String);
  end;

// This is called by the uploader with Synchronize
procedure TfrmClientMain.OnUploadError(aError: String);
begin
  // show and log the error
end;

// This is called by the uploader with Synchronize
procedure TfrmClientMain.OnCompleteFile(aFileInfo: TFileInfo);
var
  I: Integer;
begin
  I := IndexOfFile(aFileInfo.ID);
  if (I >= 0) and (I < fFiles.Count) then
  begin
    aFileInfo.Status := fsUploaded;
    aFileInfo.UploadedSize := aFileInfo.Size;
    FFiles.Items[I] := aFileInfo;
    Inc(FFilesUploaded);
    TFile.Delete(aFileInfo.Path);
    colProgressImg.UpdateCell(I);
  end;
end;

procedure TfrmClientMain.ProcessFolder;
var
  NewFiles: TStringDynArray;
  I, J: Integer;
  FileInfo: TFileInfo;
begin
    // Remove completed files from the list if it contains more than XX files
    while FFiles.Count > 1000 do
      if FFiles[0].Status = fsUploaded then
      begin
        Dec(FFilesUploaded);
        FFiles.Delete(0);
      end else
        Break;

    NewFiles := TDirectory.GetFiles(WatchFolder, ''*.*'',TSearchOption.soAllDirectories);
    for I := 0 to Length(NewFiles) - 1 do
    begin
          FileInfo.ID := FUniqueID;
          Inc(FUniqueID);
          FileInfo.Path := NewFiles[I];
          FileInfo.Size := GetFileSizeByName(NewFiles[I]);
          FileInfo.UploadedSize := 0;
          FileInfo.Status := fsToBeQueued;
          FFiles.Add(FileInfo);

      if (I mod 100) = 0 then
      begin
        UpdateStatusLabel;
        grFiles.RowCount := FFiles.Count;
        Application.ProcessMessages;
        if fUploader = nil then
          break;
      end;
    end;

    // Send the new files and resend failed to the uploader thread
    for I := 0 to FFiles.Count - 1 do
      if (FFiles[I].Status = fsToBeQueued) then
      begin
        if fUploader = nil then
          Break;
        FileInfo := FFiles[I];
        FileInfo.Status := fsQueued;
        FFiles[I] := FileInfo;
        SaveDebug(1, ''Add:    '' + ExtractFileName(FFiles[I].Path));
        FUploader.AddFile(FFiles[I]);
      end;
end;

procedure TfrmClientMain.tmrGUITimer(Sender: TObject);
var
  FileInfo: TFileInfo;
  I: Integer;
begin
  if (fUploader = nil) or not fUploader.Uploading then
    Exit;
  FileInfo := fUploader.GetProgress;
  I := IndexOfFile(FileInfo.ID);
  if (I >= 0) and (I < fFiles.Count) then
  begin
    fFiles.Items[I] := FileInfo;
    fCurrentFileName := ExtractFileName(FileInfo.Path);
    colProgressImg.UpdateCell(I);
  end;
end;

function TfrmClientMain.IndexOfFile(aID: Integer): Integer;
var
  I: Integer;
begin
  Result := -1;
  for I := 0 to FFiles.Count - 1 do
    if FFiles[I].ID = aID then
      Exit(I);
end;

Answer 1

Puede que este no sea el problema, pero TFileInfo es un registro.

Esto significa que cuando se pasa como un parámetro (no const / var), se copia. Esto puede provocar problemas con cosas como cadenas en el registro que no obtienen recuentos de referencia actualizados cuando se copia el registro.

Una cosa para intentar sería convertirlo en una clase y pasar una instancia como parámetro (es decir, un puntero a los datos en el montón).

Otra cosa a tener en cuenta es compartir Int64 (por ejemplo, sus valores de tamaño) en sistemas de 32 bits con rosca.

Actualizar / leer esto no se hace atómicamente y no tiene ninguna protección específica, por lo que es posible que una lectura del valor se diferencie de 32 bits superior e inferior debido al enhebrado. (por ejemplo, leer los 32 bits superiores, escribir los 32 bits superiores, escribir los 32 bits inferiores, leer los 32 bits inferiores, con las lecturas y escribir en diferentes hilos). Probablemente esto no esté causando los problemas que está viendo y, a menos que esté trabajando con transferencias de archivos de> 4 GB, es poco probable que le cause ningún problema.

Answer 2

Los callejones sin salida son definitivamente difíciles de detectar, pero este puede ser el problema. En su código, no vi que haya agregado tiempo de espera a la cola, al vistazo o a la eliminación de la secuencia, lo que significa que tomará el valor predeterminado de Infinite.

El enqueue tiene esta línea, lo que significa que, como cualquier objeto de sincronización, se bloqueará hasta que Enter se complete (bloquee el monitor) o se produzca el tiempo de espera (ya que no tiene un tiempo de espera, esperará por siempre)

TSimpleThreadedQueue.Enqueue(const Item: T; Timeout: LongWord): TWaitResult;
...    
if not TMonitor.Enter(FQueue, Timeout)

También voy a suponer que implementó PEEK usted mismo basado en Dequeue, solo que en realidad no elimina el artículo.

Parece que implementa su propio tiempo de espera; sin embargo, todavía tiene lo siguiente:

function TSimpleThreadedQueue.Peek/Dequeue(var Item: T; Timeout: LongWord): TWaitResult;
...
if not TMonitor.Enter(FQueue, Timeout)

Donde el tiempo de espera es infinito, entonces, si está en el método Peek esperando que se señalice con un tiempo de espera infinito, entonces no puede Enqueue algo de un segundo hilo sin bloquear ese hilo esperando que el método peek se complete en un tiempo de espera infinito

Aquí hay un fragmento del comentario de TMonitor

Enter locks the monitor object with an optional timeout (in ms) value. 
Enter without a timeout will wait until the lock is obtained. 
If the procedure returns it can be assumed that the lock was acquired. 
Enter with a timeout will return a boolean status indicating whether or 
not the lock was obtained (True) or the attempt timed out prior to 
acquire the lock (False). Calling Enter with an INFINITE timeout 
is the same as calling Enter without a timeout.

Dado que la implementación usa Infinite por defecto, y no se proporciona un valor TMonitor.Spinlock, eso bloqueará el hilo hasta que pueda adquirir el objeto FQueue.

Mi sugerencia sería cambiar tu código de la siguiente manera:

  // Use Peek instead of Dequeue, because the item should not be removed from the queue if it fails
  while true do
    case fQueue.Peek(fCurrentFile,10) 
      wrSignaled:
        try
          if UploadFile(fCurrentFile) = '''' then
          begin
            fQueue.Dequeue(Temp);  // Delete the item from the queue if succesful
            SignalComplete;
          end;
        except
          on E: Exception do
            SignalError(E.Message);
        end;
      wrTimeout: sleep(10);
      wrIOCompletion,
      wrAbandoned,
      wrError: break;
    end; //case

De esta forma, Peek no mantendrá el bloqueo en FQueue indefinidamente, dejando una ventana para que Enqueue lo adquiera y agregue el archivo del hilo principal (UI).

Answer 3

Esto podría ser una posibilidad remota, pero aquí hay otra posibilidad [la primera respuesta puede ser más probable] (algo que acabo de encontrar, pero que había sabido antes): el uso de Sincronizar puede estar causando el punto muerto. Aquí hay un blog sobre por qué sucede esto: Delphi-Workaround-for-TThread-SynchronizeWaitFor-.aspx

El punto pertinente del artículo:

El hilo A llama a Sincronizar (MethodA)

El hilo B llama a Synchronize (MethodB)

Luego, dentro del contexto del hilo principal:

El hilo principal llama a CheckSynchronize () mientras procesa mensajes

CheckSynchronize se implementa para procesar por lotes todas las llamadas en espera (*). Por lo tanto, recoge la cola de las llamadas en espera (que contienen MethodA y MethodB) y las recorre una a una.

MethodA se ejecuta en el contexto del hilo principal. Suponer MethodA llama a ThreadB.WaitFor

WaitFor llama a CheckSynchronize para procesar cualquier llamada en espera para Sincronizar

En teoría, esto debería procesar la Sincronización de ThreadB (Método B), permitiendo que se complete el Subproceso B. Sin embargo, MethodB ya es una posesión de la primera llamada CheckSynchronize, por lo que nunca se llama.

¡PUNTO MUERTO!

El artículo de Embarcadero QC describe el problema con más detalle.

Si bien no veo ninguna llamada a ProcessMessages en el código anterior, o para el caso, un WaitFor que se invocaría durante una Sincronización, podría ser un problema que en el punto en que se llama a una sincronización, otro hilo llame a la sincronización como bien, pero el hilo principal ya se ha sincronizado y está bloqueando.

Esto no hizo clic conmigo al principio, porque tiendo a evitar Sincronizar llamadas como la peste y usualmente diseñar actualizaciones de UI desde subprocesos usando otros métodos como pasar mensajes y listas de seguridad de subprocesos con notificación de mensajes en lugar de sincronizar llamadas.