Merge branch 'main' into add-sun397-datapipe

Author: YosuaMichael

.git-blame-ignore-revs

@@ -0,0 +1,7 @@
+# This file keeps git blame clean.
+# See https://docs.github.com/en/repositories/working-with-files/using-files/viewing-a-file#ignore-commits-in-the-blame-view
+
+# Add ufmt (usort + black) as code formatter (#4384)
+5f0edb97b46e5bff71dc19dedef05c5396eeaea2
+# update python syntax >=3.6 (#4585)
+d367a01a18a3ae6bee13d8be3b63fd6a581ea46f

test/builtin_dataset_mocks.py

@@ -10,6 +10,7 @@
 import pathlib
 import pickle
 import random
+import warnings
 import xml.etree.ElementTree as ET
 from collections import defaultdict, Counter
 
@@ -470,7 +471,10 @@ def imagenet(info, root, config):
         ]
         num_children = 1
         synsets.extend((0, "", "", "", num_children, [], 0, 0) for _ in range(5))
-        savemat(data_root / "meta.mat", dict(synsets=synsets))
+        with warnings.catch_warnings():
+            # The warning is not for savemat, but rather for some internals savemet is using
+            warnings.filterwarnings("ignore", category=np.VisibleDeprecationWarning)
+            savemat(data_root / "meta.mat", dict(synsets=synsets))
 
         make_tar(root, devkit_root.with_suffix(".tar.gz").name, compression="gz")
     else:  # config.split == "test"

test/test_functional_tensor.py

@@ -67,7 +67,7 @@ class TestRotate:
     IMG_W = 26
 
     @pytest.mark.parametrize("device", cpu_and_gpu())
-    @pytest.mark.parametrize("height, width", [(26, IMG_W), (32, IMG_W)])
+    @pytest.mark.parametrize("height, width", [(7, 33), (26, IMG_W), (32, IMG_W)])
     @pytest.mark.parametrize(
         "center",
         [
@@ -77,7 +77,7 @@ class TestRotate:
         ],
     )
     @pytest.mark.parametrize("dt", ALL_DTYPES)
-    @pytest.mark.parametrize("angle", range(-180, 180, 17))
+    @pytest.mark.parametrize("angle", range(-180, 180, 34))
     @pytest.mark.parametrize("expand", [True, False])
     @pytest.mark.parametrize(
         "fill",

test/test_models.py

@@ -745,24 +745,24 @@ def test_detection_model_validation(model_fn):
     x = [torch.rand(input_shape)]
 
     # validate that targets are present in training
-    with pytest.raises(ValueError):
+    with pytest.raises(AssertionError):
         model(x)
 
     # validate type
     targets = [{"boxes": 0.0}]
-    with pytest.raises(TypeError):
+    with pytest.raises(AssertionError):
         model(x, targets=targets)
 
     # validate boxes shape
     for boxes in (torch.rand((4,)), torch.rand((1, 5))):
         targets = [{"boxes": boxes}]
-        with pytest.raises(ValueError):
+        with pytest.raises(AssertionError):
             model(x, targets=targets)
 
     # validate that no degenerate boxes are present
     boxes = torch.tensor([[1, 3, 1, 4], [2, 4, 3, 4]])
     targets = [{"boxes": boxes}]
-    with pytest.raises(ValueError):
+    with pytest.raises(AssertionError):
         model(x, targets=targets)
 
 

test/test_models_detection_anchor_utils.py

@@ -16,7 +16,7 @@ def test_incorrect_anchors(self):
         image1 = torch.randn(3, 800, 800)
         image_list = ImageList(image1, [(800, 800)])
         feature_maps = [torch.randn(1, 50)]
-        pytest.raises(ValueError, anc, image_list, feature_maps)
+        pytest.raises(AssertionError, anc, image_list, feature_maps)
 
     def _init_test_anchor_generator(self):
         anchor_sizes = ((10,),)

test/test_ops.py

@@ -138,13 +138,13 @@ def test_autocast(self, x_dtype, rois_dtype):
 
     def _helper_boxes_shape(self, func):
         # test boxes as Tensor[N, 5]
-        with pytest.raises(ValueError):
+        with pytest.raises(AssertionError):
             a = torch.linspace(1, 8 * 8, 8 * 8).reshape(1, 1, 8, 8)
             boxes = torch.tensor([[0, 0, 3, 3]], dtype=a.dtype)
             func(a, boxes, output_size=(2, 2))
 
         # test boxes as List[Tensor[N, 4]]
-        with pytest.raises(ValueError):
+        with pytest.raises(AssertionError):
             a = torch.linspace(1, 8 * 8, 8 * 8).reshape(1, 1, 8, 8)
             boxes = torch.tensor([[0, 0, 3]], dtype=a.dtype)
             ops.roi_pool(a, [boxes], output_size=(2, 2))

test/test_prototype_builtin_datasets.py

@@ -7,18 +7,23 @@
 import torch
 from builtin_dataset_mocks import parametrize_dataset_mocks, DATASET_MOCKS
 from torch.testing._comparison import assert_equal, TensorLikePair, ObjectPair
-from torch.utils.data.datapipes.iter.grouping import ShardingFilterIterDataPipe as ShardingFilter
 from torch.utils.data.graph import traverse
-from torchdata.datapipes.iter import IterDataPipe, Shuffler
+from torch.utils.data.graph_settings import get_all_graph_pipes
+from torchdata.datapipes.iter import IterDataPipe, Shuffler, ShardingFilter
 from torchvision._utils import sequence_to_str
 from torchvision.prototype import transforms, datasets
+from torchvision.prototype.datasets.utils._internal import INFINITE_BUFFER_SIZE
 from torchvision.prototype.features import Image, Label
 
 assert_samples_equal = functools.partial(
     assert_equal, pair_types=(TensorLikePair, ObjectPair), rtol=0, atol=0, equal_nan=True
 )
 
 
+def extract_datapipes(dp):
+    return get_all_graph_pipes(traverse(dp, only_datapipe=True))
+
+
 @pytest.fixture
 def test_home(mocker, tmp_path):
     mocker.patch("torchvision.prototype.datasets._api.home", return_value=str(tmp_path))
@@ -35,6 +40,7 @@ def test_coverage():
         )
 
 
+@pytest.mark.filterwarnings("error")
 class TestCommon:
     @parametrize_dataset_mocks(DATASET_MOCKS)
     def test_smoke(self, test_home, dataset_mock, config):
@@ -118,19 +124,18 @@ def test_serializable(self, test_home, dataset_mock, config):
 
         pickle.dumps(dataset)
 
+    # TODO: we need to enforce not only that both a Shuffler and a ShardingFilter are part of the datapipe, but also
+    #  that the Shuffler comes before the ShardingFilter. Early commits in https://github.com/pytorch/vision/pull/5680
+    #  contain a custom test for that, but we opted to wait for a potential solution / test from torchdata for now.
     @parametrize_dataset_mocks(DATASET_MOCKS)
     @pytest.mark.parametrize("annotation_dp_type", (Shuffler, ShardingFilter))
     def test_has_annotations(self, test_home, dataset_mock, config, annotation_dp_type):
-        def scan(graph):
-            for node, sub_graph in graph.items():
-                yield node
-                yield from scan(sub_graph)
 
         dataset_mock.prepare(test_home, config)
 
         dataset = datasets.load(dataset_mock.name, **config)
 
-        if not any(type(dp) is annotation_dp_type for dp in scan(traverse(dataset))):
+        if not any(isinstance(dp, annotation_dp_type) for dp in extract_datapipes(dataset)):
             raise AssertionError(f"The dataset doesn't contain a {annotation_dp_type.__name__}() datapipe.")
 
     @parametrize_dataset_mocks(DATASET_MOCKS)
@@ -144,6 +149,17 @@ def test_save_load(self, test_home, dataset_mock, config):
             buffer.seek(0)
             assert_samples_equal(torch.load(buffer), sample)
 
+    @parametrize_dataset_mocks(DATASET_MOCKS)
+    def test_infinite_buffer_size(self, test_home, dataset_mock, config):
+        dataset_mock.prepare(test_home, config)
+        dataset = datasets.load(dataset_mock.name, **config)
+
+        for dp in extract_datapipes(dataset):
+            if hasattr(dp, "buffer_size"):
+                # TODO: replace this with the proper sentinel as soon as https://github.com/pytorch/data/issues/335 is
+                #  resolved
+                assert dp.buffer_size == INFINITE_BUFFER_SIZE
+
 
 @parametrize_dataset_mocks(DATASET_MOCKS["qmnist"])
 class TestQMNIST:

torchvision/csrc/io/decoder/decoder.cpp

@@ -18,25 +18,6 @@ constexpr size_t kIoBufferSize = 96 * 1024;
 constexpr size_t kIoPaddingSize = AV_INPUT_BUFFER_PADDING_SIZE;
 constexpr size_t kLogBufferSize = 1024;
 
-int ffmpeg_lock(void** mutex, enum AVLockOp op) {
-  std::mutex** handle = (std::mutex**)mutex;
-  switch (op) {
-    case AV_LOCK_CREATE:
-      *handle = new std::mutex();
-      break;
-    case AV_LOCK_OBTAIN:
-      (*handle)->lock();
-      break;
-    case AV_LOCK_RELEASE:
-      (*handle)->unlock();
-      break;
-    case AV_LOCK_DESTROY:
-      delete *handle;
-      break;
-  }
-  return 0;
-}
-
 bool mapFfmpegType(AVMediaType media, MediaType* type) {
   switch (media) {
     case AVMEDIA_TYPE_AUDIO:
@@ -202,8 +183,6 @@ void Decoder::initOnce() {
     avcodec_register_all();
 #endif
     avformat_network_init();
-    // register ffmpeg lock manager
-    av_lockmgr_register(&ffmpeg_lock);
     av_log_set_callback(Decoder::logFunction);
     av_log_set_level(AV_LOG_ERROR);
     VLOG(1) << "Registered ffmpeg libs";
@@ -277,7 +256,7 @@ bool Decoder::init(
           break;
       }
 
-      fmt = av_find_input_format(fmtName);
+      fmt = (AVInputFormat*)av_find_input_format(fmtName);
     }
 
     const size_t avioCtxBufferSize = kIoBufferSize;
@@ -495,8 +474,8 @@ void Decoder::cleanUp() {
 
 // function does actual work, derived class calls it in working thread
 // periodically. On success method returns 0, ENODATA on EOF, ETIMEDOUT if
-// no frames got decoded in the specified timeout time, and error on
-// unrecoverable error.
+// no frames got decoded in the specified timeout time, AVERROR_BUFFER_TOO_SMALL
+// when unable to allocate packet and error on unrecoverable error
 int Decoder::getFrame(size_t workingTimeInMs) {
   if (inRange_.none()) {
     return ENODATA;
@@ -505,10 +484,15 @@ int Decoder::getFrame(size_t workingTimeInMs) {
   // once decode() method gets called and grab some bytes
   // run this method again
   // init package
-  AVPacket avPacket;
-  av_init_packet(&avPacket);
-  avPacket.data = nullptr;
-  avPacket.size = 0;
+  // update 03/22: moving memory management to ffmpeg
+  AVPacket* avPacket;
+  avPacket = av_packet_alloc();
+  if (avPacket == nullptr) {
+    LOG(ERROR) << "decoder as not able to allocate the packet.";
+    return AVERROR_BUFFER_TOO_SMALL;
+  }
+  avPacket->data = nullptr;
+  avPacket->size = 0;
 
   auto end = std::chrono::steady_clock::now() +
       std::chrono::milliseconds(workingTimeInMs);
@@ -520,8 +504,12 @@ int Decoder::getFrame(size_t workingTimeInMs) {
   int result = 0;
   size_t decodingErrors = 0;
   bool decodedFrame = false;
-  while (!interrupted_ && inRange_.any() && !decodedFrame && watcher()) {
-    result = av_read_frame(inputCtx_, &avPacket);
+  while (!interrupted_ && inRange_.any() && !decodedFrame) {
+    if (watcher() == false) {
+      result = ETIMEDOUT;
+      break;
+    }
+    result = av_read_frame(inputCtx_, avPacket);
     if (result == AVERROR(EAGAIN)) {
       VLOG(4) << "Decoder is busy...";
       std::this_thread::yield();
@@ -538,10 +526,11 @@ int Decoder::getFrame(size_t workingTimeInMs) {
       break;
     }
 
-    // get stream
-    auto stream = findByIndex(avPacket.stream_index);
+    // get stream; if stream cannot be found reset the packet to
+    // default settings
+    auto stream = findByIndex(avPacket->stream_index);
     if (stream == nullptr || !inRange_.test(stream->getIndex())) {
-      av_packet_unref(&avPacket);
+      av_packet_unref(avPacket);
       continue;
     }
 
@@ -553,7 +542,7 @@ int Decoder::getFrame(size_t workingTimeInMs) {
       bool hasMsg = false;
       // packet either got consumed completely or not at all
       if ((result = processPacket(
-               stream, &avPacket, &gotFrame, &hasMsg, params_.fastSeek)) < 0) {
+               stream, avPacket, &gotFrame, &hasMsg, params_.fastSeek)) < 0) {
         LOG(ERROR) << "processPacket failed with code: " << result;
         break;
       }
@@ -585,20 +574,18 @@ int Decoder::getFrame(size_t workingTimeInMs) {
 
     result = 0;
 
-    av_packet_unref(&avPacket);
+    av_packet_unref(avPacket);
   }
 
-  av_packet_unref(&avPacket);
-
+  av_packet_free(&avPacket);
   VLOG(2) << "Interrupted loop"
           << ", interrupted_ " << interrupted_ << ", inRange_.any() "
           << inRange_.any() << ", decodedFrame " << decodedFrame << ", result "
           << result;
 
   // loop can be terminated, either by:
   // 1. explcitly iterrupted
-  // 2. terminated by workable timeout
-  // 3. unrecoverable error or ENODATA (end of stream)
+  // 3. unrecoverable error or ENODATA (end of stream) or ETIMEDOUT (timeout)
   // 4. decoded frames pts are out of the specified range
   // 5. success decoded frame
   if (interrupted_) {

torchvision/csrc/io/decoder/stream.cpp

@@ -28,7 +28,7 @@ Stream::~Stream() {
 
 // look up the proper CODEC querying the function
 AVCodec* Stream::findCodec(AVCodecParameters* params) {
-  return avcodec_find_decoder(params->codec_id);
+  return (AVCodec*)avcodec_find_decoder(params->codec_id);
 }
 
 // Allocate memory for the AVCodecContext, which will hold the context for

torchvision/csrc/io/decoder/subtitle_stream.cpp

@@ -43,21 +43,34 @@ int SubtitleStream::initFormat() {
 int SubtitleStream::analyzePacket(const AVPacket* packet, bool* gotFrame) {
   // clean-up
   releaseSubtitle();
+
+  // FIXME: should this even be created?
+  AVPacket* avPacket;
+  avPacket = av_packet_alloc();
+  if (avPacket == nullptr) {
+    LOG(ERROR)
+        << "decoder as not able to allocate the subtitle-specific packet.";
+    // alternative to ENOMEM
+    return AVERROR_BUFFER_TOO_SMALL;
+  }
+  avPacket->data = nullptr;
+  avPacket->size = 0;
   // check flush packet
-  AVPacket avPacket;
-  av_init_packet(&avPacket);
-  avPacket.data = nullptr;
-  avPacket.size = 0;
-  auto pkt = packet ? *packet : avPacket;
+  auto pkt = packet ? packet : avPacket;
+
   int gotFramePtr = 0;
-  int result = avcodec_decode_subtitle2(codecCtx_, &sub_, &gotFramePtr, &pkt);
+  // is these a better way than cast from const?
+  int result =
+      avcodec_decode_subtitle2(codecCtx_, &sub_, &gotFramePtr, (AVPacket*)pkt);
 
   if (result < 0) {
     LOG(ERROR) << "avcodec_decode_subtitle2 failed, err: "
                << Util::generateErrorDesc(result);
+    // free the packet we've created
+    av_packet_free(&avPacket);
     return result;
   } else if (result == 0) {
-    result = pkt.size; // discard the rest of the package
+    result = pkt->size; // discard the rest of the package
   }
 
   sub_.release = gotFramePtr;
@@ -66,9 +79,10 @@ int SubtitleStream::analyzePacket(const AVPacket* packet, bool* gotFrame) {
   // set proper pts in us
   if (gotFramePtr) {
     sub_.pts = av_rescale_q(
-        pkt.pts, inputCtx_->streams[format_.stream]->time_base, timeBaseQ);
+        pkt->pts, inputCtx_->streams[format_.stream]->time_base, timeBaseQ);
   }
 
+  av_packet_free(&avPacket);
   return result;
 }