«Сверхзвуковая» загрузка фотографий в Облако с помощью собственного NSInputStream

Максимально быстрая загрузка фотографий и видео с устройства на сервер была нашим основным приоритетом при разработке мобильного приложения Облако Mail.Ru для iOS. Кроме того, с самой первой версии приложения мы предоставили пользователям возможность включить автоматическую загрузку на сервер всего содержимого системной галереи. Это очень удобно для тех, кто волнуется о возможной потере телефона, однако, как вы понимаете, увеличивает объем передаваемых данных в разы.

Итак, мы поставили перед собой задачу сделать загрузку фото и видео из мобильного приложения Облака Mail.Ru не просто хорошей, а близкой к идеальной. Результатом стала наша библиотека POSInputStreamLibrary, которая реализует потоковую загрузку в сеть фото и видео из системной галереи iOS. Благодаря ее тесной интеграции с фреймворками ALAssetLibrary и CFNetwork загрузка в приложении происходит очень быстро и не требует ни байта свободного места на устройстве. О реализации собственного наследника класса NSInputStream из iOS Developer Library я расскажу в этом посте.

За время службы на благо Облака Mail.Ru поток POSBlobInputStream оброс весьма богатой функциональностью:

• инициализация потока URL-ом ALAsset
• поддержка синхронного и асинхронного режимов работы
• автоматическая переинициализация после инвалидации объекта ALAsset
• кеширующее чтение данных из ALAsset
• возможность указать смещение, с которого будет начато чтение
• возможность интеграции с произвольным источником данных

Смысл каждой из перечисленных возможностей разъясняется в отдельном пункте. Перед их рассмотрением осталось только сказать, что исходный код библиотеки доступен здесь, а также в главном репозитории CocoaPods.

Инициализация потока URL-ом ALAsset

До тех пор, пока вся функциональность приложения ограничивалась лишь загрузкой фотографий, все было просто. Изображение из галереи сохранялось во временный файл, на основе которого создавался стандартный файловый поток. Последний подавался на вход NSURLRequest для стриминга в сеть.

@interface NSInputStream (NSInputStreamExtensions)
// ...
+ (id)inputStreamWithFileAtPath:(NSString *)path;
// ...
@end

@interface NSMutableURLRequest (NSMutableHTTPURLRequest)
// ...
- (void)setHTTPBodyStream:(NSInputStream *)inputStream;
// ...
@end

Кликабельно:



Требование поддержать загрузку видеофайлов сделало этот подход непригодным. Огромный размер роликов порождал следующие проблемы:

• для загрузки требовалось наличие большого количества свободного места на устройстве
• время сохранения видео во временный файл могло достигать 10 и более минут

Для преодоления этих неудобств был разработан класс POSBlobInputStream. Он инициализируется URL-ом объекта галереи и читает данные напрямую без создания временных файлов.

@interface NSInputStream (POS)
+ (NSInputStream *)pos_inputStreamWithAssetURL:(NSURL *)assetURL;
+ (NSInputStream *)pos_inputStreamWithAssetURL:(NSURL *)assetURL asynchronous:(BOOL)asynchronous;
+ (NSInputStream *)pos_inputStreamForCFNetworkWithAssetURL:(NSURL *)assetURL;
@end

Кликабельно:



Поначалу у меня было ощущение, что реализация POSBlobInputStream займет минимум времени, поскольку интерфейс его базового класса тривиален.

@interface NSInputStream : NSStream
- (NSInteger)read:(uint8_t *)buffer maxLength:(NSUInteger)len;
- (BOOL)getBuffer:(uint8_t **)buffer length:(NSUInteger *)len;
- (BOOL)hasBytesAvailable;
@end

Более того, согласно документации, getBuffer:length: поддерживать необязательно, так что, казалось бы, нужно реализовать всего 2 метода. Их отображение на интерфейс ALAssetRepresentation вопросов также не вызывало.

@interface ALAssetRepresentation : NSObject
// ...
- (long long)size;
- (NSUInteger)getBytes:(uint8_t *)buffer fromOffset:(long long)offset length:(NSUInteger)length error:(NSError **)error;
// ...
@end

Однако, спустив новоиспеченный POSBlobInputStream на воду, я был неприятно удивлен. Вызов любого метода базового класса NSStream завершался исключением вида:

*** -propertyForKey: only defined for abstract class.  Define -[POSBlobInputStream propertyForKey:]

Причина заключается в том, что NSInputStream — это абстрактный класс, а каждый из его init-методов создает объект одного из классов-наследников. В Objective-C этот паттерн называется class cluster. Таким образом, реализация собственного потока требует реализации в том числе и всех методов NSStream, а их там полна горница.

@interface NSStream : NSObject
- (void)open;
- (void)close;
- (id <NSStreamDelegate>)delegate;
- (void)setDelegate:(id <NSStreamDelegate>)delegate;
- (id)propertyForKey:(NSString *)key;
- (BOOL)setProperty:(id)property forKey:(NSString *)key;
- (void)scheduleInRunLoop:(NSRunLoop *)aRunLoop forMode:(NSString *)mode;
- (void)removeFromRunLoop:(NSRunLoop *)aRunLoop forMode:(NSString *)mode;
- (NSStreamStatus)streamStatus;
- (NSError *)streamError;
@end

Синхронный и асинхронный режимы работы POSBlobInputStream

При разработке POSBlobInputStream наиболее сложным было реализовать механизм асинхронного уведомления об изменении состояния. В NSStream за него отвечают методы scheduleInRunLoop:forMode:, removeFromRunLoop:forMode: и setDelegate:. Благодаря им можно создавать такие потоки, которые на момент открытия не располагают ни байтом информации. POSBlobInputStream эксплуатирует эту возможность для следующих целей:

• Реализация неблокирующей версии метода open. POSBlobInputStream считается открытым, как только ему удалось получить объект ALAssetRepresentation по его NSURL. Как известно, с помощью iOS SDK это можно сделать только асинхронно. Таким образом, наличие механизма для асинхронного уведомления об изменении состояния потока с NSStreamStatusNotOpen на NSStreamStatusOpen или NSStreamStatusError здесь как нельзя кстати.
• Информирование о наличии у потока данных для чтения посредством отправки события NSStreamEventHasBytesAvailable.

В иллюстративных целях ниже приводятся реализации подсчета контрольной суммы файла с использованием POSBlobInputStream. Начнем с рассмотрения синхронного варианта.

NSInputStream *stream = [NSInputStream pos_inputStreamWithAssetURL:assetURL asynchronous:NO];
[stream open];
if ([stream streamStatus] == NSStreamStatusError) {
    /* Информируем об ошибке */
    return;
}
NSParameterAssert([stream streamStatus] == NSStreamStatusOpen);
while ([stream hasBytesAvailable]) {
    uint8_t buffer[kBufferSize];
    const NSInteger readCount = [stream read:buffer maxLength:kBufferSize];
    if (readCount < 0) {
        /* Информируем об ошибке */
        return;
    } else if (readCount > 0) {
        /* Логика подсчета контрольной суммы */
    }
}
if ([stream streamStatus] != NSStreamStatusAtEnd) {
    /* Информируем об ошибке */
    return;
}
[stream close];

При всей простоте у этого кода есть одна невидимая особенность. Если исполнять его в главном треде, то произойдет deadlock. Дело в том, что метод open блокирует вызывающий тред до тех пор, пока iOS SDK не вернет в главном потоке ALAsset. Если же функция open сама по себе будет вызвана в главном потоке, то получится классическая взаимоблокировка. Зачем вообще понадобилась синхронная реализация потока, будет описано ниже в разделе “Особенности интеграции с NSURLRequest”.
Асинхронная версия подсчета контрольной суммы выглядит немного сложнее.

@interface ChecksumCalculator () <NSStreamDelegate>
@end

@implementation ChecksumCalculator

- (void)calculateChecksumForStream:(NSInputStream *)aStream {
    aStream.delegate = self;
    [aStream open];
    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
            NSRunLoop *runLoop = [NSRunLoop currentRunLoop];
            [aStream scheduleInRunLoop:runLoop forMode:NSDefaultRunLoopMode];
            for (;;) { @autoreleasepool {
                if (![runLoop runMode:NSDefaultRunLoopMode
                                 beforeDate:[NSDate dateWithTimeIntervalSinceNow:kRunLoopInterval]]) {
                    break;
                }
                const NSStreamStatus streamStatus = [aStream streamStatus];
                if (streamStatus == NSStreamStatusError || streamStatus == NSStreamStatusClosed) {
                    break;
                }
            }}
    });
}

#pragma mark - NSStreamDelegate

- (void)stream:(NSStream *)aStream handleEvent:(NSStreamEvent)eventCode {
    switch (eventCode) {
        case NSStreamEventHasBytesAvailable: {
            [self updateChecksumForStream:aStream];
        } break;
        case NSStreamEventEndEncountered: {
            [self notifyChecksumCalculationCompleted];
            [_stream close];
        } break;
        case NSStreamEventErrorOccurred: {
            [self notifyErrorOccurred:[_stream streamError]];
            [_stream close];
        } break;
    }
}

@end

ChecksumCalculator устанавливает себя в качестве обработчика событий POSBlobInputStream. Как только у потока появляются новые данные, либо, наоборот, заканчиваются, либо происходит ошибка, он шлет соответствующие события. Обратите внимание, что существует возможность указать, в какой тред их слать. Например, в приведенном листинге кода они будут приходить в некий рабочий поток, созданный GCD.

Особенности интеграции с ALAssetLibrary

При работе с ALAssetLibrary следует учитывать следующее:

• Вызовы методов ALAssetRepresentation обходятся очень дорого. POSBlobInputStream старается минимизировать их количество за счет кеширования полученных результатов. Например, существует минимальный блок данных, который будет вычитан при вызове метода read:maxLength:, и только по его исчерпании произойдет новое обращение.
• ALAssetRepresentation может становиться недействительным. Так, на iOS 5.x это происходит при сохранении фотографии в галерею телефона. С точки зрения клиентского кода это выглядит как возврат нулевого значения методом getBytes:fromOffset:length:error: объекта ALAssetRepresentation. При этом заведомо известно, что данные до конца не прочитаны. В этом случае POSBlobInputStream получает ALAssetRepresentation заново. Нелишним будет отметить, что при работе в синхронном режиме на время переинициализации вызывающий поток блокируется, а в асинхронном — нет.

Особенности интеграции с NSURLRequest

В основе реализации сетевого уровня iOS SDK в целом и NSURLRequest в частности лежит фреймворк CFNetwork. За долгие годы жизни он накопил немало шкафов со скелетами. Но обо всем по порядку.

NSInputStream является одним из "toll-free bridged" классов iOS SDK. Его можно привести к CFReadStreamRef и работать с ним в дальнейшем как с объектом данного типа. Это свойство лежит в основе реализации NSURLRequest. Последний выдает POSBlobInputStream за своего брата-близнеца, и CFNetwork общается с ним уже с помощью С-интерфейса. В теории все C-вызовы к CFReadStream должны проксироваться на вызовы соответствующих им методов NSInputStream. Однако на практике есть два серьезных отклонения:

1. Не все вызовы проксируются. Для некоторых эту процедуру приходится делать самостоятельно. Останавливаться на этом здесь не буду, поскольку в интернете есть хорошие статьи на эту тему: How to implement a CoreFoundation toll-free bridget NSInputStream, Subclassing NSInputStream.
2. Проксирование CFReadStreamGetError приводит к падению приложения. Это эксклюзивное знание было получено путем анализа crash-логов приложения и медитаций над исходниками CFStream. Видимо, по этой причине указанная функция помечена в документации устаревшей, но, тем не менее, ее использование еще не искоренено изо всех мест CFNetwork. Так, каждый раз, когда NSInputStream информирует CFNetwork об ошибке, фреймворк пытается получить ее описание, используя эту злосчастную функцию. Итог печален.

Для борьбы со второй проблемой вариантов не так много. Поскольку отрефакторить CFNetwork невозможно, остается только не провоцировать его на враждебные действия. Чтобы CFNetwork не пытался получить описание ошибки, нужно ни при каких условиях не сообщать ему о ее появлении. По этой причине POSBlobInputStream обзавелся свойством shouldNotifyCoreFoundationAboutStatusChange. Если флаг выставлен, то:

1. поток не будет слать уведомления об изменении своего статуса посредством callback-ов C
2. метод streamStatus никогда не вернет значение NSStreamStatusError

Единственный способ узнать о возникновении ошибки при поднятом флаге — реализовать неким классом протокол NSStreamDelegate и установить его в качестве делегата потоку (см. пример подсчета контрольной суммы выше).

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

Поддержка чтения данных со смещением

iOS-приложение Облака Mail.Ru умеет дозагружать файлы. Данная функциональность позволяет экономить трафик и время пользователя в случае, когда часть загружаемого файла уже находится в хранилище. Для реализации этого требования POSBlobInputStream был обучен считыванию содержимого фотографии не с начала, а с некоторой позиции. Смещение в нем задается свойством NSStreamFileCurrentOffsetKey. Благодаря тому, что оно же используется для сдвига начала стандартного файлового потока, появляется возможность указывать его единообразно.

Поддержка произвольных источников данных

POSBlobInputStream был создан для загрузки фото и видео из галереи. Однако спроектирован он таким образом, чтобы в случае необходимости можно было использовать и другие источники данных. Для стриминга из других источников необходимо реализовать протокол POSBlobInputStreamDataSource.

@protocol POSBlobInputStreamDataSource <NSObject>
//
// Self-explanatory KVO-compliant properties.
@property (nonatomic, readonly, getter = isOpenCompleted) BOOL openCompleted;
@property (nonatomic, readonly) BOOL hasBytesAvailable;
@property (nonatomic, readonly, getter = isAtEnd) BOOL atEnd;
@property (nonatomic, readonly) NSError *error;
//
// This selector will be called before anything else.
- (void)open;
//
// Data Source configuring.
- (id)propertyForKey:(NSString *)key;
- (BOOL)setProperty:(id)property forKey:(NSString *)key;
//
// Data Source data.
// The contracts of these selectors are the same as for NSInputStream.
- (NSInteger)read:(uint8_t *)buffer maxLength:(NSUInteger)maxLength;
- (BOOL)getBuffer:(uint8_t **)buffer length:(NSUInteger *)bufferLength;
@end

Свойства используются не только для получения состояния источника данных, но и для информирования потока о его изменении с помощью механизма KVO.

Итог

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

#pragma mark Undocumented but necessary NSStream Overrides (fuck you Apple)

- (void) _scheduleInCFRunLoop:(NSRunLoop*) inRunLoop forMode:(id)inMode {
    /* FUCK YOU APPLE */
}

- (void) _setCFClientFlags:(CFOptionFlags)inFlags
                  callback:(CFReadStreamClientCallBack)inCallback
                   context:(CFStreamClientContext)inContext {
    /* NO SERIOUSLY, FUCK YOU */
}

POSBlobInputStream, в свою очередь, является одним из ключевых компонентов приложения Облака Mail.Ru. За время службы он был проверен в бою армией пользователей. Было собрано и нивелировано множество граблей, и в данный момент поток является одним из наиболее стабильных компонентов. Пользуйтесь, пишите расширения, и, конечно, буду рад любой обратной связи.

Павел Осипов,
Руководитель команды разработки Облака для iOS