class Job<TInput, TOutput>
{
    private readonly TInput data;
    private readonly TaskCompletionSource<TOutput> taskSource;

    public Job(TInput data)
    {
        this.data = data;
        this.taskSource = new TaskCompletionSource<TOutput>();
    }

    public Task<TOutput> Task
    {
        get
        {
            return this.taskSource.Task;
        }
    }

    public async void RunAsync()
    {
        try
        {
            var result = await DoRealJobAsync(this.data);
            this.taskSource.SetResult(result);
        }
        catch (OperationCanceledException)
        {
            this.taskSource.TrySetCanceled();
        }
        catch (Exception x)
        {
            this.taskSource.TrySetException(x);
        }
    }

    private static async Task<TOutput> DoRealJobAsync(TInput data)
    {
        await ...
        ...
        await ...
    }

}

class JobScheduler<TInput, TOutput> : IDisposable
{
    private readonly ConcurrentQueue<Job> jobQueue;
    private readonly Timer timer;
    private readonly TimeSpan runPeriod;
    private readonly int numConcurrentJobs;

    public Fetcher(int numConcurrentJobs, TimeSpan runPeriod)
    {
        this.numConcurrentJobs = numConcurrentJobs;
        this.runPeriod = runPeriod;

        this.jobQueue = new ConcurrentQueue<Job>();
        this.timer = new Timer(RunJobs);

        this.timer.Change(TimeSpan.Zero, this.runPeriod);
    }

    public Task<TOutput> ScheduleJobAsync(TInput data)
    {
        var job = new Job<TInput, TOutput>(data);
        this.jobQueue.Enqueue(job);
        return job.Task;
    }

    public Dispose()
    {
        this.timer.Dispose();
    }

    private void RunJobs(object state)
    {
        for (var i = 0; i < this.numConcurrentJobs; ++i)
        {
            Job job;
            if (this.jobQueue.TryDequeue(out job))
                job.RunAsync();
        }
    }
}

class Job<TInput, TOutput>
{
    private readonly TInput data;
    private readonly TaskCompletionSource<TOutput> taskSource;

    public Job(TInput data)
    {
        this.data = data;
        this.taskSource = new TaskCompletionSource<TOutput>();
    }

    public Task<TOutput> Task
    {
        get
        {
            return this.taskSource.Task;
        }
    }

    public async void RunAsync()
    {
        try
        {
            var result = await DoRealJobAsync(this.data);
            this.taskSource.SetResult(result);
        }
        catch (OperationCanceledException)
        {
            this.taskSource.TrySetCanceled();
        }
        catch (Exception x)
        {
            this.taskSource.TrySetException(x);
        }
    }

    private static async Task<TOutput> DoRealJobAsync(TInput data)
    {
        await ...
        ...
        await ...
    }

}

class JobScheduler<TInput, TOutput> : IDisposable
{
    private readonly ConcurrentQueue<Job> jobQueue;
    private readonly Timer timer;
    private readonly TimeSpan runPeriod;
    private readonly int numConcurrentJobs;

    public JobScheduler(int numConcurrentJobs, TimeSpan runPeriod)
    {
        this.numConcurrentJobs = numConcurrentJobs;
        this.runPeriod = runPeriod;

        this.jobQueue = new ConcurrentQueue<Job>();
        this.timer = new Timer(RunJobs);

        this.timer.Change(TimeSpan.Zero, this.runPeriod);
    }

    public Task<TOutput> ScheduleJobAsync(TInput data)
    {
        var job = new Job<TInput, TOutput>(data);
        this.jobQueue.Enqueue(job);
        return job.Task;
    }

    public Dispose()
    {
        this.timer.Dispose();
    }

    private void RunJobs(object state)
    {
        for (var i = 0; i < this.numConcurrentJobs; ++i)
        {
            Job job;
            if (this.jobQueue.TryDequeue(out job))
                job.RunAsync();
        }
    }
}