Same thread instead of concurrency

Question

5.00/5 (2 votes)

See more:

Hi all,

I'm afraid it's my (messed up) syntax that results in a repeated task running syncroniously instead of using multiple tasks (i.e. the 'Debug.WriteLine' message reports Thread 5 all the time).

Would someone who has the relevant experience please have a look and guide me to the proper syntax? I've edited out all the comments and irrelevant things, hoping it's well readable for you guys.

Thanks a lot,
Mick

What I have tried:

I've (of course) tried various patterns found on the internet, but get more confused with each one of them. My current code basically looks like that:

Call from another (async) method:

VB

Dim tasks = txtLists.Select(Function(x) TestForLTO_async(x))
Dim results = Await Task.WhenAll(tasks)

VB

Private Async Function TestForLTO_async(testFile As FileInfo) As Task(Of FileInfo)
     Dim isLTO = Await FilterForLTO(testFile)
     If isLTO Then ListFilesLTO.Add(testFile)
     Return If(isLTO, testFile, Nothing)
End Function

Private Function FilterForLTO(datei As FileInfo) As Task(Of Boolean)
     Dim isLTO As Boolean = False
     ' True if first lines match regex
     Dim content As String = String.Join(", ", File.ReadLines(datei.FullName).Take(3))
     If regex_SerialNumber.IsMatch(content) Then
          isLTO = True
          Debug.WriteLine($"{datei.Name} is of LTO-Type (Thread ID {Threading.Thread.CurrentThread.ManagedThreadId})")
     end if
     Return Task.FromResult(isLTO)
End Function

Posted 4-Aug-23 22:48pm

Sonhospa

Updated 4-Aug-23 23:00pm

v2

Add a Solution

1 solution

Add a Solution

This content, along with any associated source code and files, is licensed under The Code Project Open License (CPOL)

Graeme_Grant · Accepted Answer · 2023-08-04T23:00:00

There are no asynchronous operations.

You need to use asynchronous File methods. Alternatively, wrap your call to TestForLTO_async in a Task.Run in the Select Linq method.

UPDATE

I have knocked up an example of how to do it with the Asynchronous file method. I do not have your files, so I modified the filtering.

VB.NET

Imports System
Imports System.IO
Imports System.Text

Module Program
    Sub Main(args As String())
        
        RunTest().GetAwaiter().GetResult()
        Console.ReadKey()

    End Sub

    Private Async Function RunTest() As Task
        
        dim files As List(Of String) =
            Directory
                .GetFiles(Environment.CurrentDirectory)
                .ToList()

        dim Infos = files.Select(Function(file) new FileInfo(file))
        
        dim tasks = Infos.Select(function(fileInfo) 
            ProcessFileAsync(fileInfo))

        dim results = Await Task.WhenAll(tasks)

        for Each fileInfo as FileInfo in results
            Console.WriteLine($"> {fileInfo.Name}")
        Next
        
    End Function

    Private Async Function ProcessFileAsync(fileInfo as FileInfo)
        As Task(Of FileInfo)
        
        if (await FilterFileAsync(fileInfo)) Then
            return fileInfo
        End If
        
        return Nothing
        
    End Function

    Private async Function FilterFileAsync(fileInfo as FileInfo)
        As Task(Of boolean)
        
        dim bytes As Byte() = await file
            .ReadAllBytesAsync(fileInfo.FullName) 
        
        Console.WriteLine($"Thread ID: 
            {Environment.CurrentManagedThreadId}")

        return (Encoding.Default.GetString(bytes).Contains("a"))
        
    End Function
    
End Module

Note: lines are wrapped to fit area. If you create a console app and drop in the code (with the wrapping fixed), you can run and see it work.

And here is the output (for me):

Thread ID: 9
Thread ID: 7
Thread ID: 8
Thread ID: 5
Thread ID: 10
> AsyncManyFilesVb
> AsyncManyFilesVb.deps.json
> AsyncManyFilesVb.runtimeconfig.json
> AsyncManyFilesVb.dll
> AsyncManyFilesVb.pdb

You can see each file is processed on a different thread.

UPDATE #2

Here is a FileStream version that should work with .Net Framework (see documentation via the link in the conversation below):

VB

Private Async Function ProcessFileAsync(fileInfo as FileInfo)
    As Task(Of FileInfo)
    
    if (await FilterFileStreamAsync(fileInfo)) Then
        return fileInfo
    End If
    
    return Nothing
    
End Function

Private async Function FilterFileStreamAsync(fileInfo as FileInfo)
    As Task(Of boolean)
    
    ' forcing a yield so that we can get the thread id before the 
    ' async file read
    await Task.Yield()
    
    Console.WriteLine($"Thread ID:
        {Environment.CurrentManagedThreadId}")
    
    using ms as MemoryStream = New MemoryStream()
        using fileStream as FileStream =
            file.Open(fileinfo.FullName, FileMode.Open)
            
            ' load the MemoryStream from the file
            await fileStream.CopyToAsync(ms)
            
            ' move back to the start of the stream
            ms.Position = 0
            Dim sr As New StreamReader(ms)
            return (sr.ReadToEnd().Contains("a"c))
            
        End Using
    End Using
    
End Function

NOTE: I've put the Thread ID output before the async call. To force the compiler to correctly report the Thread ID I use a Task.Yield(). I've done this just in case you need to do it before and not after. If you don't, it will report the same Thread ID. Comment out the Task.Yield() and run the code and you will see the same Thread ID. A trap for many. ;)

And here is the output:

Thread ID: 10
Thread ID: 8
Thread ID: 7
Thread ID: 9
Thread ID: 5
> AsyncManyFilesVb
> AsyncManyFilesVb.deps.json
> AsyncManyFilesVb.runtimeconfig.json
> AsyncManyFilesVb.dll
> AsyncManyFilesVb.pdb

UPDATE #3

I woke up this morning with your question on my mind. I would do things a little different. So, for educational purposes, here is how I would modify my answer for my own use. The aim is to:
1. simplify the API and expose a custom filter that can be changed when calling the processor
2. apply the Task result handler to use what was learned in the last question: Tasks not being created?[^]

For #2, we can convert to an extension method:

VB

Module TaskExtensions

    <System.Runtime.CompilerServices.Extension>
    Public Iterator Function WaitForAllCompleted(Of T) _
        (tasks As IList(Of Task(Of T))) _
        As IEnumerable(Of Task(Of T))

        While tasks.Count > 0

            Dim completed As List(Of Task(Of T)) =
                    tasks.Where(Function(task) task.IsCompleted).ToList()

            If completed?.Any() = True Then

                For Each task As Task In completed.ToList()
                    Yield task
                    tasks.Remove(task)
                Next

            End If

        End While

    End Function

End Module

Next, for #1, I would move the processing to a service Class or Module (static class). It depends on how you want to use it.

VB

Module SomeService

    Public Iterator Function ProcessFiles(
        files As IEnumerable(Of FileInfo),
        filter As Predicate(Of (file As FileInfo, fileData As String))) _
        As IEnumerable(Of Task(Of FileInfo))

        For Each file As FileInfo In files

            Dim result = ProcessFileAsync(file, filter)
            Yield result
        Next

    End Function

    Private Async Function ProcessFileAsync(fileInfo As FileInfo,
        filter As Predicate(Of (file As FileInfo, fileData As String))) _
        As Task(Of FileInfo)

        If (Await FilterFileStreamAsync(fileInfo, filter)) Then
            Return fileInfo
        End If

        Return Nothing

    End Function

    Private Async Function FilterFileStreamAsync(fileInfo As FileInfo,
        filter As Predicate(Of (file As FileInfo, fileData As String))) _
        As Task(Of Boolean)

        ' forcing a yield so that we can get the thread id before the 
        ' async file read
        Await Task.Yield()

        Console.WriteLine($"* Thread ID:
           {Environment.CurrentManagedThreadId}")

        Using ms As MemoryStream = New MemoryStream()
            Using fileStream As FileStream =
                File.Open(fileInfo.FullName, FileMode.Open)

                ' load the MemoryStream from the file
                Await fileStream.CopyToAsync(ms)

                ' move back to the start of the stream
                ms.Position = 0
                Dim sr As New StreamReader(ms)

                'return the filtered result
                Return filter((fileInfo, Await sr.ReadToEndAsync()))

            End Using
        End Using

    End Function

End Module

Here I am using a Predicate(Of T) Delegate [^] to pass a filter into the service method. TLDR; a Predicate returns a Boolean value. I am also using a Yield Statement[^] to return each item for handling rather than the complete list. The remainder is Identical to my previous answer above (update #2).

Now we can use the new service API and Task extension method:

VB.NET

Imports System.IO

Module Module1

    Sub Main(args As String())

        Console.WriteLine($"-- started: {DateTime.Now}")

        RunTest().GetAwaiter().GetResult()

        Console.WriteLine($"-- Finished: {DateTime.Now}")
        Console.WriteLine()
        Console.WriteLine($"- press any key to exit -")
        Console.ReadKey()

    End Sub

    Private Async Function RunTest() As Task

        Dim files As List(Of String) = Directory _
                .GetFiles(Environment.CurrentDirectory) _
                .Where(Function(file) _
                          Not (Path.GetExtension(file).Equals(".exe") OrElse
                               Path.GetExtension(file).Equals(".pdb"))) _
                .ToList()

        Dim Infos = files.Select(Function(file) New FileInfo(file))

        Dim filter As Predicate(Of (file As FileInfo, fileData As String)) =
                Function(data) data.fileData.Contains("a"c)

        For Each completedTask In SomeService.ProcessFiles(Infos, filter) _
            .ToList().WaitForAllCompleted()

            Dim fileInfo = Await completedTask

            If fileInfo IsNot Nothing Then
                Console.WriteLine($"> {fileInfo.Name}")
            End If

        Next

    End Function

End Module

As you can see, I am defining a filter, then passing it to the API call SomeService.ProcessFiles(Infos, filter), I convert that to a list so we can remove completed tasks, then calling the Task extension method WaitForAllCompleted. The whole statement is used in a For Each ... as we Yield each completed task rather than the whole list of tasks allowing for further processing or reporting as each Task completes, rather than at the end of the process.

And yes, this was written and tested on my Dev (Windows) PC. That is why I am filtering out the .exe and .pdb files.

The code is a little bit more verbose however now we have moved the different parts to their own concerns (single responsibilities), exposes a simple API, allows custom filtering, and is now easier to maintain and use.

Lastly, the output:

-- started: 6/08/2023 8:54:38 AM
* Thread ID: 3
* Thread ID: 4
> AsyncManyFilesVb.xml
> AsyncManyFilesVb.exe.config
-- Finished: 6/08/2023 8:54:38 AM

- press any key to exit -

Lots to unpack.... enjoy!

Same thread instead of concurrency

1 solution

Solution 1

Add your solution here

Preview 0

Same thread instead of concurrency

1 solution

Solution 1

Add your solution here

Preview 0

Existing Members

...or Join us