Reescrevendo foreach usando a estrutura IOBServable e reativa

Question

Estou no VS2008 com a estrutura da entidade. Estou acessando objetos do banco de dados usando o ESQL para onde na funcionalidade. Estou passando uma tonelada de IDs para a instrução SELECT, então eu a coloquei em conjuntos de 800. Em seguida, fundi os resultados de cada pedaço. Meu objetivo é obter resultados para cada pedaço em paralelo, em vez de esperar de maneira síncrona. Instalei a estrutura reativa e tenho certeza de que preciso usar o Forkjoin. No entanto, não consigo descobrir como converter essa função para usá -la. Aqui está o meu código existente:

    public static IList<TElement> SelectWhereIn<TElement, TValue>(this ObjectContext context, string fieldName, IList<TValue> idList)
    {
        var chunkedIds = idList.Split(CHUNK_SIZE);
        string entitySetName = typeof(TElement).Name + "Set";
        var retList = new List<TElement>();
        foreach (var idChunk in chunkedIds)
        {
            string delimChunk = string.Join(",", idChunk.Select(x => x.ToString()).ToArray());
            ObjectQuery<TElement> query = context.CreateQuery<TElement>("SELECT VALUE x FROM " + entitySetName + " AS x");
            query = query.Where("it." + fieldName + " IN {" + delimChunk + "}");
            retList.AddRange(query);
        }
        return retList;
    }

Obrigado!

Editar >>> Modifiquei o código para usar os homens pobres como abaixo:

    public static IList<TElement> SelectWhereIn<TElement, TValue>(this ObjectContext context, string fieldName, IList<TValue> idList)
    {
        var chunkedIds = idList.Split(CHUNK_SIZE);
        string entitySetName = typeof(TElement).Name + "Set";
        var chunkLists = new List<IEnumerable<TElement>>();
        Parallel.ForEach(chunkedIds, idChunk =>
        {
            string delimChunk = string.Join(",", idChunk.Select(x => x.ToString()).ToArray());
            ObjectQuery<TElement> query = context.CreateQuery<TElement>("SELECT VALUE x FROM " + entitySetName + " AS x");
            query = query.Where("it." + fieldName + " IN {" + delimChunk + "}");
            chunkLists.Add(query.ToList());
        });
        var retList = new List<TElement>();
        foreach (var chunkList in chunkLists)
        {
            retList.AddRange(chunkList);
        }
        return retList;
    }

Funcionou muito bem pela primeira vez. Mas a segunda vez que o corri, recebi este erro:

A conexão não foi fechada. O estado atual da conexão está se conectando. Descrição: ocorreu uma exceção não tratada durante a execução da solicitação da Web atual. Revise o rastreamento da pilha para obter mais informações sobre o erro e onde se originou no código.

Detalhes da exceção: System.InValidoPerationException: A conexão não foi fechada. O estado atual da conexão está se conectando.

Erro de fonte:

Linha 49: foreach (var iasyncresult na lista de resultados) Linha 50: {linha 51: del.endinvoke (IIAsyncresult); Linha 52: iasyncresult.asyncwaithandle.close (); Linha 53:}

É interessante, b/c Emre (o autor da biblioteca) tem uma edição em seu post original falando sobre como ele adicionou essas linhas de código para maior segurança. Estou usando certo? Ou seu V1 estava mais seguro, afinal?

Answer 1

O VS2010 tem isso com Plinq. Usando as extensões AsParallel().WithDegreeOfParallelism(nbProcessors) faria o que você precisa.

Com VS2008, eu usei Paralelo do pobre homem. No passado, quando eu estava tentando contornar um gargalo de desempenho, tente tentar.

EDIT: Em reação ao erro que você adicionou, pode ser um tiro aleatório nos contextos escuros, mas separados para cada thread? Igual a:

Parallel.ForEach(chunkedIds, idChunk =>
    {
        ObjectContext context = new MyContext(connStr);//depending what's your config
                                                       // like, with or w/o conn string

        string delimChunk = string.Join(",", idChunk.Select(x => x.ToString()).ToArray());
        ObjectQuery<TElement> query = context.CreateQuery<TElement>("SELECT VALUE x FROM " + entitySetName + " AS x");
        query = query.Where("it." + fieldName + " IN {" + delimChunk + "}");
        chunkLists.Add(query.ToList());
    });

Você pode precisar ajustar algumas coisas (como pegue a string de conexão do contexto estendido para instanciar novos contextos).