Dali: How to return ops.Uniform() as Tensor

[vinkle]

class RandomRotatedSimplePipeline(Pipeline):
    def __init__(self, batch_size, num_threads, device_id):
        super(RandomRotatedSimplePipeline, self).__init__(batch_size, num_threads, device_id, seed = 12)
        self.input = ops.FileReader(file_root = image_dir, random_shuffle = True, initial_fill = 21)
        self.decode = ops.ImageDecoder(device = 'cpu', output_type = types.RGB)
        self.rotate = ops.Rotate()
        self.rng = ops.Uniform(range = (-10.0, 10.0))

    def define_graph(self):
        jpegs, labels = self.input()
        images = self.decode(jpegs)
        angle = self.rng()
        rotated_images = self.rotate(images, angle = angle)
        return (rotated_images, labels, angle) 

if I return angle, it gives following errors.

self._pipe.Build(self._names_and_devices)
RuntimeError: [/opt/dali/dali/pipeline/executor/executor.h:638] Assert on "!tensor.producer.is_support" failed: Outputs of support ops cannot be outputs.

[awolant]

Hi, thanks for the question.
This is due to some limitations in support ops. We are in the process of reworking them into regular CPU ops (e.g. #1259). Then this inconvenience will disappear.

[vinkle]

Great, thanks for your reply.
Is there any immediate workaround to get tensors as outputs. Also, how much time do you think will it take to become part of pre-built installation package?

[awolant]

Immediate solution that comes to my mind is to write a custom op that would allow argument input and copy it to its output. Then you could return output of this custom op as the pipeline output.

It is hard to give an ETA for this at the moment, but it will not happen in the next release.

[vinkle]

Hi thanks for your reply.
I was able to compile a custom operator and it has the same job as to copy the content to output array. But, I am still getting errors. Can you please look into it.
This is my custome operator code (which is same as the example code from the dali documentation)

// supportop_to_tensor.h
#ifndef SUPPORTOP_TO_TENSOR
#define SUPPORTOP_TO_TENSOR

#include <vector>
#include "dali/pipeline/operators/operator.h"

namespace other_ns 
{
    template <typename Backend>
    class SupportOp2Tensor : public ::dali::Operator<Backend> 
    {
        public:
        inline explicit SupportOp2Tensor(const ::dali::OpSpec &spec) :
            ::dali::Operator<Backend>(spec) {}

        virtual inline ~SupportOp2Tensor() = default;

        SupportOp2Tensor(const SupportOp2Tensor&) = delete;
        SupportOp2Tensor& operator=(const SupportOp2Tensor&) = delete;
        SupportOp2Tensor(SupportOp2Tensor&&) = delete;
        SupportOp2Tensor& operator=(SupportOp2Tensor&&) = delete;

        protected:
        bool CanInferOutputs() const override 
        {
            return true;
        }

        bool SetupImpl(std::vector<::dali::OutputDesc> &output_desc,
                       const ::dali::workspace_t<Backend> &ws) override 
        {
            const auto &input = ws.template InputRef<Backend>(0);
            output_desc.resize(1);
            output_desc[0] = {input.shape(), input.type()};
            return true;
        }
        void RunImpl(::dali::Workspace<Backend> *ws) override;
    };
}
#endif  

// supportop_to_tensor.cc
#include "supportop_to_tensor.h"
namespace other_ns 
{
  template<>
  void SupportOp2Tensor<::dali::CPUBackend>::RunImpl(::dali::SampleWorkspace *ws) 
  {
    const auto &input = ws->Input<::dali::CPUBackend>(0);
    auto &output = ws->Output<::dali::CPUBackend>(0);

    ::dali::TypeInfo type = input.type();
    type.Copy<::dali::CPUBackend, ::dali::CPUBackend>(
        output.raw_mutable_data(),
        input.raw_data(), input.size(), 0);
  }
}
DALI_REGISTER_OPERATOR(SupportOp2Tensor, ::other_ns::SupportOp2Tensor<::dali::CPUBackend>, ::dali::CPU);
DALI_SCHEMA(SupportOp2Tensor)
  .DocStr("Returns the support op (eg. ops.Uniform) as the tensor")
  .NumInput(1)
  .NumOutput(1);

And this is a python code to call the custom op

class RandomRotatedSimplePipeline(Pipeline):
    def __init__(self, batch_size, num_threads, device_id):
        super(RandomRotatedSimplePipeline, self).__init__(batch_size, num_threads, device_id, seed = 12)
        self.input = ops.FileReader(file_root = image_dir, random_shuffle = True, initial_fill = 21)
        self.decode = ops.ImageDecoder(device = 'cpu', output_type = types.RGB)
        self.rotate = ops.Rotate()
        self.rng = ops.Uniform(range = (-10.0, 10.0))
        # custome op to return support op as a tensor
        self.rng2tensor= ops.SupportOp2Tensor()

    def define_graph(self):
        jpegs, labels = self.input()
        images = self.decode(jpegs)
        angle = self.rng()
        rotated_images = self.rotate(images, angle = angle)
        angle_tensor = self.rng2tensor(angle)
        return (rotated_images, labels, angle_tensor) 

This is the error I am getting.

 in _prepare_graph
    related_logical_id[op.relation_id] = self._pipe.AddOperator(op.spec, op.name)
RuntimeError: [/opt/dali/dali/pipeline/pipeline.cc:263] Assert on "!it->second.is_support" failed: Argument input can only be used as regular input by support ops. (op: 'SupportOp2Tensor', input: 'Uniform_id_2_output_0')
Stacktrace (22 entries):

Can you please help me getting fix the error ?

Thanks

[JanuszL]

Hi,
To get input from support ops you need to thread it as an argument, not as the regular input.
So it could be like this (based on https://github.com/NVIDIA/DALI/blob/master/dali/pipeline/operators/fused/crop_mirror_normalize.h#L214):

// you need to add optional argument that accepts input form the support op
DALI_REGISTER_OPERATOR(SupportOp2Tensor, ::other_ns::SupportOp2Tensor<::dali::CPUBackend>, ::dali::CPU);
DALI_SCHEMA(SupportOp2Tensor)
  .DocStr("Returns the support op (eg. ops.Uniform) as the tensorlist")
  .NumInput(0)
  .NumOutput(1)
  .AddOptionalArg("support_op_data",
    R"code(some support op data)code",  0, true);

template<>
  void SupportOp2Tensor<::dali::CPUBackend>::RunImpl(::dali::SampleWorkspace *ws) 
  {
    auto &output = ws->Output<::dali::CPUBackend>(0);
    for (int data_idx = 0; data_idx < batch_size_; data_idx++) {
        input.raw_data()[data_idx ] = this->spec_.template GetArgument<int>("support_op_data", &ws, data_idx);
    }
  }

[vinkle]

Thanks, I was able to compile using the suggested changes. However, I am still getting an error during graph compilation.

ValueError: Operator SupportOp2Tensor expects [0, 0] inputs, but received 1

Is it because we have set the .NumInput(0) in the DALI_SCHEMA?
I have changed a bit RunImpl definition since I was not able to assign values to output.raw_data.

  void SupportOp2Tensor<::dali::CPUBackend>::RunImpl(::dali::SampleWorkspace *ws) {
    auto &output = ws->Output<::dali::CPUBackend>(0);
    std::vector<float>_temp(batch_size_);
    for (int data_idx = 0; data_idx < batch_size_; data_idx++){
      auto &output = ws->Output<::dali::CPUBackend>(0);
      _temp[data_idx] = spec_.template GetArgument<float>("support_op_data", ws, data_idx);
    }
    ::dali::TypeInfo type = output.type();
    type.Copy<::dali::CPUBackend, ::dali::CPUBackend>(
      output.raw_mutable_data(),
      &_temp[0], _temp.size(), 0);
  }

[JanuszL]

I think in your pipeline you should have something like:

    def define_graph(self):
        (...)
        out_angel = self.SupportOp2Tensor(support_op_data = angle)
        (...)
        return (rotated_images, labels, out_angel)

[vinkle]

Great, thank you very much for your help. I was able to get the output of support op as a tensorlist :) . I have only one little issue that the returned tensor is of size batch_size_xbatch_size_ instead of batch_size_x1 and all the rows are duplicate of the first row.
e.g for batch_size=4:

angles=
[[-6.927826   6.17914   -4.365263  -3.4307795]
 [-6.927826   6.17914   -4.365263  -3.4307795]
 [-6.927826   6.17914   -4.365263  -3.4307795]
 [-6.927826   6.17914   -4.365263  -3.4307795]]

Below is a code for your reference.

//supportop_to_tensor.h
#ifndef SUPPORTOP_TO_TENSOR
#define SUPPORTOP_TO_TENSOR
#include <vector>
#include "dali/pipeline/operators/operator.h"

namespace other_ns {
    template <typename Backend>
    class SupportOp2Tensor : public ::dali::Operator<Backend> {
        public:
        inline explicit SupportOp2Tensor(const ::dali::OpSpec &spec) :
            ::dali::Operator<Backend>(spec) {}
        virtual inline ~SupportOp2Tensor() = default;
        SupportOp2Tensor(const SupportOp2Tensor&) = delete;
        SupportOp2Tensor& operator=(const SupportOp2Tensor&) = delete;
        SupportOp2Tensor(SupportOp2Tensor&&) = delete;
        SupportOp2Tensor& operator=(SupportOp2Tensor&&) = delete;
        protected:
        bool CanInferOutputs() const override {
            return false;
        }
        bool SetupImpl(std::vector<::dali::OutputDesc> &output_desc,
                       const ::dali::workspace_t<Backend> &ws) override {
            return false;
        }
        protected:
        void RunImpl(::dali::Workspace<Backend> *ws) override;
    };
}
#endif  

//supportop_to_tensor.cc
#include "supportop_to_tensor.h"
namespace other_ns {
  template<>
  void SupportOp2Tensor<::dali::CPUBackend>::RunImpl(::dali::SampleWorkspace *ws) {
    auto &output = ws->Output<::dali::CPUBackend>(0);
    output.Resize({batch_size_});
    float *_t = output.mutable_data<float>();
    for (int data_idx = 0; data_idx < batch_size_; data_idx++){
      _t[data_idx] = spec_.template GetArgument<float>("support_op_data", ws, data_idx);
    }
  }
}
DALI_REGISTER_OPERATOR(SupportOp2Tensor, ::other_ns::SupportOp2Tensor<::dali::CPUBackend>, ::dali::CPU);
DALI_SCHEMA(SupportOp2Tensor)
  .DocStr("Returns the support op (eg. ops.Uniform) as the tensorlist")
  .NumInput(0)
  .NumOutput(1)
  .AddOptionalArg("support_op_data", R"code(some support op data)code",  0, true);

[JanuszL]

My bad. I have forgot that you are writing CPU op that works on sample basis. So SupportOp2Tensor<::dali::CPUBackend>::RunImpl is invoked per every sample.
What you should do:

template<>
  void SupportOp2Tensor<::dali::CPUBackend>::RunImpl(::dali::SampleWorkspace *ws) {
    auto &output = ws->Output<::dali::CPUBackend>(0);
    output.Resize({1});
    float *_t = output.mutable_data<float>();
    int data_idx = ws.data_idx();
    _t[data_idx] = spec_.template GetArgument<float>("support_op_data", ws, data_idx);
  }

Or even:

//supportop_to_tensor.h
#ifndef SUPPORTOP_TO_TENSOR
#define SUPPORTOP_TO_TENSOR
#include <vector>
#include "dali/pipeline/operators/operator.h"

namespace other_ns {
    template <typename Backend>
    class SupportOp2Tensor : public ::dali::Operator<Backend> {
        public:
        inline explicit SupportOp2Tensor(const ::dali::OpSpec &spec) :
            ::dali::Operator<Backend>(spec) {}
        virtual inline ~SupportOp2Tensor() = default;
        SupportOp2Tensor(const SupportOp2Tensor&) = delete;
        SupportOp2Tensor& operator=(const SupportOp2Tensor&) = delete;
        SupportOp2Tensor(SupportOp2Tensor&&) = delete;
        SupportOp2Tensor& operator=(SupportOp2Tensor&&) = delete;
        protected:
        bool CanInferOutputs() const override {
            return false;
        }
        bool SetupImpl(std::vector<::dali::OutputDesc> &output_desc,
                       const ::dali::workspace_t<Backend> &ws) override {
            return false;
        }
        protected:
        void RunImpl(::dali::workspace_t<Backend> *ws) override;
    };
}
#endif 

//supportop_to_tensor.cc
#include "supportop_to_tensor.h"
namespace other_ns {
  template<>
  void SupportOp2Tensor<::dali::CPUBackend>::RunImpl(::dali::HostWorkspace *ws) {
    for (int data_idx = 0; data_idx < batch_size_; data_idx++){
      auto &output = ws->Output<::dali::CPUBackend>(0, data_idx);
      output.Resize({1});
      float *_t = output.mutable_data<float>();
      _t = spec_.template GetArgument<float>("support_op_data", ws, data_idx);
    }
  }
}
DALI_REGISTER_OPERATOR(SupportOp2Tensor, ::other_ns::SupportOp2Tensor<::dali::CPUBackend>, ::dali::CPU);
DALI_SCHEMA(SupportOp2Tensor)
  .DocStr("Returns the support op (eg. ops.Uniform) as the tensorlist")
  .NumInput(0)
  .NumOutput(1)
  .AddOptionalArg("support_op_data", R"code(some support op data)code",  0, true);

[vinkle]

You folks are great. Thank you very much :)
Just a minor update to a pointer error (_t should be *_t).
*_t = spec_.template GetArgument<float>("support_op_data", ws, data_idx);

I am updating the complete program who wants replicate.

//supportop_to_tensor.h
#ifndef SUPPORTOP_TO_TENSOR
#define SUPPORTOP_TO_TENSOR
#include "dali/pipeline/operators/operator.h"
namespace other_ns {
    template <typename Backend>
    class SupportOp2Tensor : public ::dali::Operator<Backend> {
        public:
        inline explicit SupportOp2Tensor(const ::dali::OpSpec &spec) :
            ::dali::Operator<Backend>(spec) {}
        virtual inline ~SupportOp2Tensor() = default;
        SupportOp2Tensor(const SupportOp2Tensor&) = delete;
        SupportOp2Tensor& operator=(const SupportOp2Tensor&) = delete;
        SupportOp2Tensor(SupportOp2Tensor&&) = delete;
        SupportOp2Tensor& operator=(SupportOp2Tensor&&) = delete;
        protected:
        bool CanInferOutputs() const override {
            return false;
        }
        bool SetupImpl(std::vector<::dali::OutputDesc> &output_desc,
                       const ::dali::workspace_t<Backend> &ws) override {
            return false;
        }
        protected:
        void RunImpl(::dali::workspace_t<Backend> *ws) override;
    };
}
#endif 

//supportop_to_tensor.cc
#include "supportop_to_tensor.h"
namespace other_ns {
  template<>
  void SupportOp2Tensor<::dali::CPUBackend>::RunImpl(::dali::HostWorkspace *ws) {
    for (int data_idx = 0; data_idx < batch_size_; data_idx++){
      auto &output = ws->Output<::dali::CPUBackend>(0, data_idx);
      output.Resize({1});
      float *_t = output.mutable_data<float>();
      *_t = (spec_.template GetArgument<float>("support_op_data", ws, data_idx));
    }
  }
}
DALI_REGISTER_OPERATOR(SupportOp2Tensor, ::other_ns::SupportOp2Tensor<::dali::CPUBackend>, ::dali::CPU);
DALI_SCHEMA(SupportOp2Tensor)
  .DocStr("Returns the support op (eg. ops.Uniform) as the tensorlist")
  .NumInput(0)
  .NumOutput(1)
  .AddOptionalArg("support_op_data", R"code(some support op data)code",  0, true);

And the python program

import numpy as np
from   nvidia.dali.pipeline import Pipeline
import nvidia.dali.ops as ops
import nvidia.dali.types as types
import nvidia.dali.plugin_manager as plugin_manager
IMAGE_DIR = "path/to/image/dir"
OPS_PATH="path/to/libsupportop2tensor.so"
plugin_manager.load_library(OPS_PATH)

batch_size = 8

class RandomRotatedSimplePipeline(Pipeline):
    def __init__(self, batch_size, num_threads, device_id):
        super(RandomRotatedSimplePipeline, self).__init__(batch_size, num_threads, device_id, seed = 12)
        self.input = ops.FileReader(file_root = IMAGE_DIR, random_shuffle = True, initial_fill = 21)
        self.decode = ops.ImageDecoder(device = 'cpu', output_type = types.RGB)
        self.rotate = ops.Rotate()
        self.rng = ops.Uniform(range = (-10.0, 10.0))
        # custom op to return support op as a tensor a tensor list
        self.rng2tensor= ops.SupportOp2Tensor()

    def define_graph(self):
        jpegs, labels = self.input()
        images = self.decode(jpegs)
        angle = self.rng()
        rotated_images = self.rotate(images, angle = angle)
        angle_tensor = self.rng2tensor(support_op_data=angle)
        return (rotated_images, labels, angle_tensor) 


pipe = RandomRotatedSimplePipeline(batch_size, 1, 0)
pipe.build()
pipe_out = pipe.run()
images, labels, angle_tensor = pipe_out
print('angles:', np.array(angle_tensor.as_tensor()))

[JanuszL]

:+1:

[JanuszL]

@vinkle We are happy to announce that latest master (after #1423) removes the distinction between support operators and cpu operators. You can now use support operators as outputs, python operator inputs, etc - and you can pass regular CPU operator outputs to named inputs.

[vinkle]

That's great. I just checked the nighly-build, and it works.
Thank you :)