преобразование блокировки потока в блокировку потока не в f #

Question

Я получаю данные из API Bloomberg, и меня очень удивляет медлительность.Мое вычисление - это ввод-вывод, ограниченный этим.

Поэтому я решил использовать какой-нибудь асинхронный монадный конструктор, чтобы отменить его.После его запуска результаты не намного лучше, что было очевидно, поскольку я выполняю вызов функции NextEvent, которая блокирует потоки.

     let outerloop args dic = 
        ...
        let rec innerloop continuetoloop   =
           let eventObj = session.NextEvent();  //This blocks
            ...

     let seqtable = reader.ReadFile( @"C:\homeware\sector.csv", ";".[0], true) 

     let dic = ConcurrentDictionary<_,_> ()
     let wf = seqtable |> Seq.mapi (fun i item -> async { outerloop item dic } )
     wf  |> Async.Parallel
         |> Async.RunSynchronously
         |> ignore
     printfn "%A" ret

Есть ли хороший способ обернуть этот блокирующий вызовна неблокирующий звонок?Кроме того, почему асинхронная среда не создает столько потоков, сколько у меня запросов (например, 200)?когда я проверяю потоки, из которых я получаю значения, я вижу только 4-5, которые используются ..

ОБНОВЛЕНИЕ

Я нашел убедительную причину того, что он никогда не будетбыть возможнымАсинхронная операция берет то, что находится после асинхронной инструкции, и планирует это где-то в пуле потоков.для всего, что имеет значение, если асинхронная функция используется правильно, то есть всегда возвращаться к пулу потоков, из которого она возникла, мы можем считать, что выполняемся в одном потоке.

Нахождение в одном потоке означаетвсе это планирование всегда будет выполняться где-то позже, и инструкция по блокировке не может избежать того факта, что, в конце концов, после выполнения ей в какой-то момент в будущем придется заблокировать поток операций.

Answer 1

Есть ли хороший способ превратить этот блокирующий вызов в неблокирующий вызов?

Нет.Вы никогда не можете обернуть блокирующие вызовы, чтобы сделать их неблокирующими.Если бы вы могли, асинхронность была бы просто шаблоном проектирования, а не фундаментальным изменением парадигмы.

Кроме того, почему асинхронная среда не создает столько потоков, сколько у меня запросов (например, 200)?

Async построен на пуле потоков, который разработан, чтобы не создавать потоки агрессивно, потому что они очень дороги.Весь смысл (реального) асинхронного заключается в том, что он не требует столько потоков, сколько имеется соединений.Вы можете обрабатывать 10 000 одновременных соединений с ~ 30 потоками.

Похоже, вы совершенно не понимаете, что такое асинхронность и для чего она нужна.Я бы посоветовал купить любую книгу по F #, которая освещает эту тему, и почитать об этом.В частности, ваше решение не является асинхронным, потому что вы просто вызываете блокирующий элемент StartGetFieldsValue из своего асинхронного рабочего процесса, что побеждает цель асинхронности.Вы могли бы просто сделать Array.Parallel.map getFieldsValue вместо этого.

Кроме того, вы хотите использовать чисто функциональный API при параллельном выполнении операций, а не мутировать ConcurrentDictionary на месте.Поэтому замените req.StartGetFieldsValue ret на

let! result = req.StartGetFieldsValue()
...
return result

и замените ignore на dict.

Answer 2

Вот решение, которое я принял, которое, кажется, работает. Конечно, он использует не только асинхронный (минус а), но и асинхронный.

Я определяю простой тип, который имеет одно событие завершения и один метод, asyncstart, с методом для запуска в качестве аргумента. он запускает метод, затем запускает событие, завершенное в соответствующем месте (в моем случае мне пришлось захватить контекст синхронизации и т. д.)

Тогда на стороне потребителя я использую просто

let! completion = Async.Waitfromevent wapper.finished |> Async.StartAsChild
let! completed = completion

Во время выполнения этого кода на стороне потребителя я использую только асинхронные вызовы, поэтому мой код не блокируется. Конечно, где-то должен быть какой-то поток, который заблокирован, но это происходит вне моего основного цикла обслуживания, который остается реактивным и пригодным.