Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Remove MakeContiguous before CPU inputs of GPU ops. #5590

Merged
merged 5 commits into from
Aug 2, 2024
Merged

Remove MakeContiguous before CPU inputs of GPU ops. #5590

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f657d6b
Remove MakeContiguous before CPU inputs of GPU ops. Remove redundant …
mzient Aug 1, 2024
8c6e9e1
Remove outdated tests.
mzient Aug 2, 2024
1524201
Bugfix.
mzient Aug 2, 2024
b59ae31
Restore test.
mzient Aug 2, 2024
7b85316
Remove more input setup.
mzient Aug 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 0 additions & 18 deletions dali/pipeline/executor/op_graph_verifier.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,24 +59,6 @@ struct parent_constraints<OpType::GPU> {
static constexpr bool supports_argument_inputs = true;
};

template <typename T, size_t N>
static constexpr bool is_in_array(const T value, const T (&arr)[N], size_t idx) {
// we encountered the end of array -> false
// otherwise we found at current position or search in next position recursivelly
return idx < N && (value == arr[idx] || is_in_array(value, arr, idx + 1));
}

template <OpType op_type>
static constexpr bool allows_op_input(OpType parent) {
return is_in_array(parent, parent_constraints<op_type>::allowed_input_ops, 0);
}

template <OpType op_type>
static constexpr bool allows_tensor_input(StorageDevice device) {
return is_in_array(device, parent_constraints<op_type>::allowed_input_tensors, 0);
}


DLL_PUBLIC std::vector<int> ArgumentInputConstraints();
DLL_PUBLIC std::vector<std::set<OpType>> ParentOpTypeConstraints();

Expand Down
12 changes: 3 additions & 9 deletions dali/pipeline/executor/workspace_policy.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,24 +53,18 @@ inline std::ostream &operator<<(std::ostream &os, StorageDevice device) {
}
}

// We instantiate the operation of adding the input only for parent op_type and device
// that are specifically allowed

// We always use queue_idx = 0 if give queue has only one element -> it is not queued
template <OpType op_type, OpType producer_type, StorageDevice device>
enable_if_t<allows_op_input<op_type>(producer_type) && allows_tensor_input<op_type>(device)>
add_input(Workspace &ws, const tensor_data_store_queue_t &storage,
int queue_idx = 0) {
void add_input(Workspace &ws, const tensor_data_store_queue_t &storage,
int queue_idx = 0) {
auto &queue = get_queue<producer_type, device>(storage);
DALI_ENFORCE(!queue.IsBuffered() || queue_idx < static_cast<int>(queue.size()),
"Backing Tensor store queue has not enough elements.");
auto tensor = queue[queue_idx];
ws.AddInput(tensor);
}

// If parent op_type or device is not allowed this is a no-op
template <OpType op_type, OpType producer_type, StorageDevice device>
enable_if_t<!allows_op_input<op_type>(producer_type) || !allows_tensor_input<op_type>(device)>
add_input(Workspace &, const tensor_data_store_queue_t &, int = 0) {}

template <OpType op_type, StorageDevice device>
void add_output(Workspace &ws, const tensor_data_store_queue_t &storage,
Expand Down
21 changes: 12 additions & 9 deletions dali/pipeline/pipeline.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -355,16 +355,19 @@ int Pipeline::AddOperatorImpl(const OpSpec &const_spec, const std::string &inst_
DALI_ENFORCE(device_id_ != CPU_ONLY_DEVICE_ID || device == "cpu",
"Cannot add a Mixed operator with a GPU output, 'device_id' "
"should not be `CPU_ONLY_DEVICE_ID`.");
} else if (input_device == "cpu") {
// device == gpu
// TODO(michalz): Add a D2H copy
DALI_ENFORCE(it->second.has_cpu,
make_string("Error while specifying ", FormatInput(spec, i),
". CPU input requested by operator exists only on GPU. CPU "
"operator cannot follow GPU operator."));
SetupCPUInput(it, i, &spec);
} else {
}

if (input_device == "gpu") {
SetupGPUInput(it);
} else {
if (device != "gpu") {
// device == gpu
// TODO(michalz): Add a D2H copy instead
DALI_ENFORCE(it->second.has_cpu,
make_string("Error while specifying ", FormatInput(spec, i),
". CPU input requested by operator exists only on GPU. CPU "
"operator cannot follow GPU operator."));
}
}
}

Expand Down
41 changes: 11 additions & 30 deletions dali/pipeline/pipeline_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ class PipelineTest : public DALITest {

// Inputs must be know to the pipeline, i.e. ops
// must be added in a topological ordering.
ASSERT_THROW(
EXPECT_THROW(
pipe.AddOperator(
OpSpec("Copy")
.AddArg("device", dev1)
Expand All @@ -71,25 +71,10 @@ class PipelineTest : public DALITest {
.AddArg("device", "gpu")
.AddOutput("data", "gpu"));

// TODO(michalz): Remove this constraint and the tests. This should be a build-time error,
// with old executor, not a construction-time error.

// For dev1 = "cpu": Inputs to CPU ops must be on CPU,
// we do not auto-copy them from gpu to cpu.
// For dev1 = "gpu": CPU inputs to GPU ops must be on CPU,
// we will not copy them back to the host.
ASSERT_THROW(
pipe.AddOperator(
OpSpec("Copy")
.AddArg("device", dev1)
.AddInput("data", dev2)
.AddOutput("copy_out", dev1)),
std::runtime_error);

if (dev1 == "cpu") {
// Inputs to CPU ops must already exist on CPU,
// we do not auto-copy them from gpu to cpu.
ASSERT_THROW(
EXPECT_THROW(
pipe.AddOperator(
OpSpec("Copy")
.AddArg("device", dev1)
Expand All @@ -109,7 +94,7 @@ class PipelineTest : public DALITest {
.AddOutput("data_3", dev1));

// Outputs must have unique names.
ASSERT_THROW(
EXPECT_THROW(
pipe.AddOperator(
OpSpec("Copy")
.AddArg("device", dev1)
Expand All @@ -125,7 +110,7 @@ class PipelineTest : public DALITest {
}
// All data must have unique names regardless
// of the device they exist on.
ASSERT_THROW(
EXPECT_THROW(
pipe.AddOperator(
OpSpec("Copy")
.AddArg("device", dev1)
Expand All @@ -135,7 +120,7 @@ class PipelineTest : public DALITest {


// CPU ops can only produce CPU outputs
ASSERT_THROW(
EXPECT_THROW(
pipe.AddOperator(
OpSpec("Copy")
.AddArg("device", dev1)
Expand Down Expand Up @@ -232,7 +217,7 @@ TYPED_TEST_SUITE(PipelineTest, NumThreads);
TEST_F(PipelineTestOnce, TestInputNotKnown) {
Pipeline pipe(1, 1, 0);

ASSERT_THROW(
EXPECT_THROW(
pipe.AddOperator(
OpSpec("Copy")
.AddArg("device", "cpu")
Expand All @@ -249,10 +234,6 @@ TEST_F(PipelineTestOnce, TestEnforceGPUOpConstraints) {
RunTestEnforce("gpu", "cpu");
}

TEST_F(PipelineTestOnce, TestTriggerToContiguous) {
RunTestTrigger("cpu");
}

Comment on lines -252 to -255
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test checked for the presence of the feature we've just removed.

TEST_F(PipelineTestOnce, TestTriggerCopyToDevice) {
RunTestTrigger("gpu");
}
Expand Down Expand Up @@ -565,7 +546,7 @@ class PrefetchedPipelineTest : public DALITest {
TEST_F(PrefetchedPipelineTest, SetQueueSizesSeparatedFail) {
Pipeline pipe(this->batch_size_, 4, 0);
// By default we are non-separated execution
ASSERT_THROW(pipe.SetQueueSizes(5, 3), std::runtime_error);
EXPECT_THROW(pipe.SetQueueSizes(5, 3), std::runtime_error);
}

TEST_F(PrefetchedPipelineTest, SetExecutionTypesFailAfterBuild) {
Expand All @@ -578,7 +559,7 @@ TEST_F(PrefetchedPipelineTest, SetExecutionTypesFailAfterBuild) {

vector<std::pair<string, string>> outputs = {{"final_images", "gpu"}};
pipe.Build(outputs);
ASSERT_THROW(pipe.SetExecutionTypes(), std::runtime_error);
EXPECT_THROW(pipe.SetExecutionTypes(), std::runtime_error);
}

TEST_F(PrefetchedPipelineTest, SetQueueSizesFailAfterBuild) {
Expand All @@ -591,7 +572,7 @@ TEST_F(PrefetchedPipelineTest, SetQueueSizesFailAfterBuild) {

vector<std::pair<string, string>> outputs = {{"final_images", "gpu"}};
pipe.Build(outputs);
ASSERT_THROW(pipe.SetQueueSizes(2, 2), std::runtime_error);
EXPECT_THROW(pipe.SetQueueSizes(2, 2), std::runtime_error);
}

TEST_F(PrefetchedPipelineTest, TestFillQueues) {
Expand Down Expand Up @@ -713,7 +694,7 @@ TEST(PipelineTest, AddOperator) {
.AddOutput("data_out1", "cpu"), "second_op", first_op);
EXPECT_EQ(first_op, second_op);

ASSERT_THROW(pipe.AddOperator(OpSpec("Copy"), "another_op", first_op), std::runtime_error);
EXPECT_THROW(pipe.AddOperator(OpSpec("Copy"), "another_op", first_op), std::runtime_error);

int third_op = pipe.AddOperator(OpSpec("DummyOpToAdd")
.AddArg("device", "cpu")
Expand All @@ -728,7 +709,7 @@ TEST(PipelineTest, AddOperator) {
.AddInput("data_in0", "cpu")
.AddOutput("data_out3", "cpu"), "DummyOpNoSync");

ASSERT_THROW(pipe.AddOperator(OpSpec("DummyOpNoSync")
EXPECT_THROW(pipe.AddOperator(OpSpec("DummyOpNoSync")
.AddArg("device", "cpu")
.AddInput("data_in0", "cpu")
.AddOutput("data_out4", "cpu"), "DummyOpNoSync2", disallow_sync_op), std::runtime_error);
Expand Down
6 changes: 4 additions & 2 deletions dali/test/python/test_pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -524,7 +524,7 @@ def define_graph(self):
)
return (output, self.labels)

for i in range(50):
for i in range(10):
pipe = HybridPipe(batch_size=batch_size, num_threads=2, device_id=0)
pipe.build()
pipe_out = pipe.run()
Expand Down Expand Up @@ -636,7 +636,9 @@ def define_graph(self):
new_pipe = Pipeline(batch_size=batch_size, num_threads=2, device_id=0)
new_pipe.deserialize_and_build(serialized_pipeline)

compare_pipelines(pipe, new_pipe, batch_size, 50)
#compare_pipelines(pipe, new_pipe, batch_size, 50)
pipe.build()
pipe.run()


def test_warpaffine():
Expand Down
Loading