<commit_before>using System;
using System.Linq;
using System.Net.Http;
using System.IO;
using System.Text;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive;
using System.Reactive.Disposables;
using Punchclock;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Net;
using Splat;
using System.Threading;

#if WINRT
using Windows.Security.Cryptography.Core;
using System.Runtime.InteropServices.WindowsRuntime;
#else
using System.Security.Cryptography;
#endif

namespace Akavache.Http
{
    class HttpCacheEntry
    {
        public string ETag { get; set; }
        public DateTimeOffset? LastModified { get; set; }
        public bool ShouldCheckResponseHeaders { get; set; }
        public HttpStatusCode Code { get; set; }

        public Dictionary<string, List<string>> Headers { get; set; }
        public Dictionary<string, List<string>> ContentHeaders { get; set; }
        public byte[] Data { get; set; }

        public bool UseCachedData(HttpResponseMessage response)
        {
            if (response.Headers.ETag != null && ETag != response.Headers.ETag.Tag)
            {
                return false;
            }

            if (response.Content.Headers.LastModified != null && LastModified != null && response.Content.Headers.LastModified > LastModified)
            {
                return false;
            }

            return true;
        }

        public static HttpCacheEntry FromResponse(HttpResponseMessage message, byte[] data)
        {
            return new HttpCacheEntry() {
                ETag = message.Headers.ETag != null ? message.Headers.ETag.Tag : null,
                LastModified = message.Content.Headers.LastModified,
                Code = message.StatusCode,
                Headers = message.Headers.ToDictionary(x => x.Key, x => x.Value.ToList()),
                ContentHeaders = message.Content.Headers.ToDictionary(x => x.Key, x => x.Value.ToList()),
                Data = data,
            };
        }

        public HttpResponseMessage ToResponse()
        {
            var ret = new HttpResponseMessage(Code)
            {
                Content = new ByteArrayContent(Data),
            };

            foreach (var kvp in Headers) { ret.Headers.TryAddWithoutValidation(kvp.Key, kvp.Value); }
            foreach (var kvp in ContentHeaders) { ret.Content.Headers.TryAddWithoutValidation(kvp.Key, kvp.Value); }
            return ret;
        }
    }

    public class CachingHttpScheduler : IHttpScheduler
    {
        readonly IBlobCache blobCache = null;
        readonly IHttpScheduler innerScheduler = null;

        readonly ConcurrentDictionary<string, IObservable<Tuple<HttpResponseMessage, byte[]>>> inflightDictionary = 
            new ConcurrentDictionary<string, IObservable<Tuple<HttpResponseMessage, byte[]>>>();

        public CachingHttpScheduler(IHttpScheduler innerScheduler = null, IBlobCache blobCache = null)
        {
            this.blobCache = blobCache;
            this.innerScheduler = innerScheduler ?? new HttpScheduler();
            
            // Things that are interesting:
            // - If-None-Match
            // - If-Modified-Since
            // - If-Range
            // - If-Unmodified-Since
            // - Last-Modified
            // - HEAD
            // - Age
            // - Cache-Control request
            // - Cache-Control response
            // - ETag
            // - ServerDate
            // - Expires
            // - Pragma: no-cache
            // - Retry-After + 503

            // Things that are tricky
            // - High prio rqs that satisfy a low prio pending request, we need
            //   to cancel / dequeue the low prio underlying one and return the high-prio
            //   one. (i.e. priority inversion)
        }

        public HttpClient Client
        {
            get { return innerScheduler.Client; }
            set { innerScheduler.Client = value; }
        }

        protected IBlobCache Cache
        {
            // NB: If we didn't look this up here, it would be impossible to
            // override the cache that Akavache.Http uses since it would be
            // saved off in the static ctor
            get { return blobCache ?? BlobCache.LocalMachine; }
        }

        public IObservable<Tuple<HttpResponseMessage, byte[]>> Schedule(HttpRequestMessage request, int priority, Func<HttpResponseMessage, bool> shouldFetchContent)
        {
            var key = UniqueKeyForRequest(request);
            var cache = default(IObservable<Tuple<HttpResponseMessage, byte[]>>);

            if (inflightDictionary.TryGetValue(key, out cache))
            {
                return cache;
            }

            if (request.Method != HttpMethod.Get || 
                (request.Headers.CacheControl != null && request.Headers.CacheControl.NoStore))
            {
                var noCache = Observable.Defer(() => innerScheduler.Schedule(request, priority, x => shouldFetchContent(x)))
                    .Finally(() => inflightDictionary.TryRemove(key, out cache))
                    .PublishLast();

                noCache.Connect();
                return noCache;
            }

            var ret = blobCache.GetObjectAsync<HttpCacheEntry>(key).Catch<HttpCacheEntry>(Observable.Return(default(HttpCacheEntry)))
                .SelectMany(async (cacheEntry, ct) =>
                {
                    var cancelSignal = new AsyncSubject<Unit>();
                    ct.Register(() => { cancelSignal.OnNext(Unit.Default); cancelSignal.OnCompleted(); });

                    if (cacheEntry != null && !cacheEntry.ShouldCheckResponseHeaders)
                    {
                        return Tuple.Create(cacheEntry.ToResponse(), cacheEntry.Data);
                    }

                    var toConcat = Observable.Empty<Tuple<HttpResponseMessage, byte[]>>();
                    var cacheIsValid = false;

                    ct.ThrowIfCancellationRequested();
                    var respWithData = await innerScheduler.Schedule(request, priority, respHeaders =>
                    {
                        if (!shouldFetchContent(respHeaders))
                        {
                            cancelSignal.OnNext(Unit.Default);
                            cancelSignal.OnCompleted();
                            return false;
                        }

                        if (cacheEntry != null && cacheEntry.UseCachedData(respHeaders))
                        {
                            toConcat = Observable.Return(Tuple.Create(cacheEntry.ToResponse(), cacheEntry.Data));
                            cacheIsValid = true;
                            return false;
                        }

                        return true;
                    }).TakeUntil(cancelSignal).Concat(Observable.Defer(() => toConcat));

                    if (cancelSignal.IsCompleted || ct.IsCancellationRequested)
                    {
                        throw new OperationCanceledException();
                    }

                    if (DefinitelyShouldntCache(respWithData.Item1) || cacheIsValid)
                    {
                        cancelSignal.OnNext(Unit.Default);
                        cancelSignal.OnCompleted();
                        return respWithData;
                    }

                    var expiryDate = CacheUntilDate(respWithData.Item1);
                    var entry = HttpCacheEntry.FromResponse(respWithData.Item1, respWithData.Item2);
                    entry.ShouldCheckResponseHeaders = (expiryDate == null);

                    await blobCache.InsertObject(key, entry, expiryDate);
                    return respWithData;
                })
                .Finally(() => inflightDictionary.TryRemove(key, out cache))
                .PublishLast();

            ret.Connect();

            inflightDictionary.TryAdd(key, ret);
            return ret;
        }

        public void ResetLimit(long? maxBytesToRead)
        {
            innerScheduler.ResetLimit(maxBytesToRead);
        }

        public void CancelAll()
        {
            innerScheduler.CancelAll();
        }

        static bool DefinitelyShouldntCache(HttpResponseMessage message)
        {
            if (!message.IsSuccessStatusCode) return true;
            if (message.Headers.CacheControl != null && message.Headers.CacheControl.NoStore) return true;
<commit_msg>Fix priority inversion in caching scheduler<commit_after>using System;
using System.Linq;
using System.Net.Http;
using System.IO;
using System.Text;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive;
using System.Reactive.Disposables;
using Punchclock;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Net;
using Splat;
using System.Threading;

#if WINRT
using Windows.Security.Cryptography.Core;
using System.Runtime.InteropServices.WindowsRuntime;
#else
using System.Security.Cryptography;
#endif

namespace Akavache.Http
{
    class HttpCacheEntry
    {
        public string ETag { get; set; }
        public DateTimeOffset? LastModified { get; set; }
        public bool ShouldCheckResponseHeaders { get; set; }
        public HttpStatusCode Code { get; set; }

        public Dictionary<string, List<string>> Headers { get; set; }
        public Dictionary<string, List<string>> ContentHeaders { get; set; }
        public byte[] Data { get; set; }

        public bool UseCachedData(HttpResponseMessage response)
        {
            if (response.Headers.ETag != null && ETag != response.Headers.ETag.Tag)
            {
                return false;
            }

            if (response.Content.Headers.LastModified != null && LastModified != null && response.Content.Headers.LastModified > LastModified)
            {
                return false;
            }

            return true;
        }

        public static HttpCacheEntry FromResponse(HttpResponseMessage message, byte[] data)
        {
            return new HttpCacheEntry() {
                ETag = message.Headers.ETag != null ? message.Headers.ETag.Tag : null,
                LastModified = message.Content.Headers.LastModified,
                Code = message.StatusCode,
                Headers = message.Headers.ToDictionary(x => x.Key, x => x.Value.ToList()),
                ContentHeaders = message.Content.Headers.ToDictionary(x => x.Key, x => x.Value.ToList()),
                Data = data,
            };
        }

        public HttpResponseMessage ToResponse()
        {
            var ret = new HttpResponseMessage(Code)
            {
                Content = new ByteArrayContent(Data),
            };

            foreach (var kvp in Headers) { ret.Headers.TryAddWithoutValidation(kvp.Key, kvp.Value); }
            foreach (var kvp in ContentHeaders) { ret.Content.Headers.TryAddWithoutValidation(kvp.Key, kvp.Value); }
            return ret;
        }
    }

    public class CachingHttpScheduler : IHttpScheduler
    {
        readonly IBlobCache blobCache = null;
        readonly IHttpScheduler innerScheduler = null;

        readonly ConcurrentDictionary<Tuple<string, int>, IObservable<Tuple<HttpResponseMessage, byte[]>>> inflightDictionary = 
            new ConcurrentDictionary<Tuple<string, int>, IObservable<Tuple<HttpResponseMessage, byte[]>>>();

        public CachingHttpScheduler(IHttpScheduler innerScheduler = null, IBlobCache blobCache = null)
        {
            this.blobCache = blobCache;
            this.innerScheduler = innerScheduler ?? new HttpScheduler();
            
            // Things that are interesting:
            // - If-None-Match
            // - If-Modified-Since
            // - If-Range
            // - If-Unmodified-Since
            // - Last-Modified
            // - HEAD
            // - Age
            // - Cache-Control request
            // - Cache-Control response
            // - ETag
            // - ServerDate
            // - Expires
            // - Pragma: no-cache
            // - Retry-After + 503

            // Things that are tricky
            // - High prio rqs that satisfy a low prio pending request, we need
            //   to cancel / dequeue the low prio underlying one and return the high-prio
            //   one. (i.e. priority inversion) (FIXED)
        }

        public HttpClient Client
        {
            get { return innerScheduler.Client; }
            set { innerScheduler.Client = value; }
        }

        protected IBlobCache Cache
        {
            // NB: If we didn't look this up here, it would be impossible to
            // override the cache that Akavache.Http uses since it would be
            // saved off in the static ctor
            get { return blobCache ?? BlobCache.LocalMachine; }
        }

        public IObservable<Tuple<HttpResponseMessage, byte[]>> Schedule(HttpRequestMessage request, int priority, Func<HttpResponseMessage, bool> shouldFetchContent)
        {
            var key = Tuple.Create(UniqueKeyForRequest(request), priority);
            var cache = default(IObservable<Tuple<HttpResponseMessage, byte[]>>);

            if (inflightDictionary.TryGetValue(key, out cache))
            {
                return cache;
            }

            if (request.Method != HttpMethod.Get || 
                (request.Headers.CacheControl != null && request.Headers.CacheControl.NoStore))
            {
                var noCache = Observable.Defer(() => innerScheduler.Schedule(request, priority, x => shouldFetchContent(x)))
                    .Finally(() => inflightDictionary.TryRemove(key, out cache))
                    .PublishLast();

                noCache.Connect();
                return noCache;
            }

            var ret = blobCache.GetObjectAsync<HttpCacheEntry>(key.Item1).Catch<HttpCacheEntry>(Observable.Return(default(HttpCacheEntry)))
                .SelectMany(async (cacheEntry, ct) =>
                {
                    var cancelSignal = new AsyncSubject<Unit>();
                    ct.Register(() => { cancelSignal.OnNext(Unit.Default); cancelSignal.OnCompleted(); });

                    if (cacheEntry != null && !cacheEntry.ShouldCheckResponseHeaders)
                    {
                        return Tuple.Create(cacheEntry.ToResponse(), cacheEntry.Data);
                    }

                    var toConcat = Observable.Empty<Tuple<HttpResponseMessage, byte[]>>();
                    var cacheIsValid = false;

                    ct.ThrowIfCancellationRequested();
                    var respWithData = await innerScheduler.Schedule(request, priority, respHeaders =>
                    {
                        if (!shouldFetchContent(respHeaders))
                        {
                            cancelSignal.OnNext(Unit.Default);
                            cancelSignal.OnCompleted();
                            return false;
                        }

                        if (cacheEntry != null && cacheEntry.UseCachedData(respHeaders))
                        {
                            toConcat = Observable.Return(Tuple.Create(cacheEntry.ToResponse(), cacheEntry.Data));
                            cacheIsValid = true;
                            return false;
                        }

                        return true;
                    }).TakeUntil(cancelSignal).Concat(Observable.Defer(() => toConcat));

                    if (cancelSignal.IsCompleted || ct.IsCancellationRequested)
                    {
                        throw new OperationCanceledException();
                    }

                    if (DefinitelyShouldntCache(respWithData.Item1) || cacheIsValid)
                    {
                        cancelSignal.OnNext(Unit.Default);
                        cancelSignal.OnCompleted();
                        return respWithData;
                    }

                    var expiryDate = CacheUntilDate(respWithData.Item1);
                    var entry = HttpCacheEntry.FromResponse(respWithData.Item1, respWithData.Item2);
                    entry.ShouldCheckResponseHeaders = (expiryDate == null);

                    await blobCache.InsertObject(key.Item1, entry, expiryDate);
                    return respWithData;
                })
                .Finally(() => inflightDictionary.TryRemove(key, out cache))
                .PublishLast();

            ret.Connect();

            inflightDictionary.TryAdd(key, ret);
            return ret;
        }

        public void ResetLimit(long? maxBytesToRead)
        {
            innerScheduler.ResetLimit(maxBytesToRead);
        }

        public void CancelAll()
        {
            innerScheduler.CancelAll();
        }

        static bool DefinitelyShouldntCache(HttpResponseMessage message)
        {
            if (!message.IsSuccessStatusCode) return true;
            if (message.Headers.CacheControl != null && message.Headers.CacheControl.NoStore) return true;

{'fraction_non_alphanumeric': 0.09102654451491661, 'fraction_numerical': 0.0009396288466055908, 'mean_word_length': 1.6703779206523444, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3196321', 'n_tokens_mistral': 4219, 'n_tokens_neox': 4013, 'n_words': 917}