首页 > 数据库 > 数据库开发技术 > 一个算子在深度学习框架中的旅程

import oneflow as torch class TinyModel(torch.nn.Module):
def __init__(self):
super(TinyModel, self).__init__()
self.linear1 = torch.nn.Linear(100, 200)
self.activation = torch.nn.ReLU()
self.linear2 = torch.nn.Linear(200, 10)
self.softmax = torch.nn.Softmax()
def forward(self, x):
x = self.linear1(x)
x = self.activation(x)
x = self.linear2(x)
x = self.softmax(x)
return xtinymodel = TinyModel()print('The model:')print(tinymodel)
output = flow.relu(input)为例,梳理一个op从Python -> C++执行的完整过程。
oneflow/api/python/framework/tensor.cpp
oneflow/api/python/framework/tensor_functions.cpp
oneflow/core/functional/functional_api.yaml
https://docs.python.org/zh-cn/3.8/c-api/index.html
https://pybind11.readthedocs.io/en/stable/index.html
python/oneflow/__init__.py
- name: "relu"signature: "Tensor (Tensor x, Bool inplace=False) => Relu"bind_python: True
tools/functional/generate_functional_api.py
build/oneflow/core/functional/functional_api.yaml.h
build/oneflow/core/functional/functional_api.yaml.cpp
class ReluFunctor {public:ReluFunctor() { op_ = CHECK_JUST(one::OpBuilder("relu").Input("x", 1).Output("y", 1).Build()); }Maybe<Tensor> operator()(const std::shared_ptr<Tensor>& x, bool inplace) const {...}private:std::shared_ptr<OpExpr> op_;};
ONEFLOW_FUNCTION_LIBRARY(m) {m.add_functor<impl::ReluFunctor>("Relu");...}
class ReluFunctor {public:ReluFunctor() { op_ = CHECK_JUST(one::OpBuilder("relu").Input("x", 1).Output("y", 1).Build()); }Maybe<Tensor> operator()(const std::shared_ptr<Tensor>& x, bool inplace) const {if (inplace) {JUST(CheckInplaceValid(x));std::shared_ptr<TensorTuple> outputs = std::make_shared<TensorTuple>(1);outputs->at(0) = x;JUST(OpInterpUtil::Dispatch(*op_, {x}, outputs.get(), AttrMap{}));return outputs->at(0);} else {return OpInterpUtil::Dispatch<Tensor>(*op_, {x});}}private:std::shared_ptr<OpExpr> op_;};
std::shared_ptr<OpExpr> op_;
class OpInterpUtil {public:template<typename T>static Maybe<T> Dispatch(const OpExpr& op_expr, const TensorTuple& inputs, const AttrMap& attrs) {return Dispatch<T>(op_expr, inputs, OpExprInterpContext(attrs));}template<typename T>static Maybe<T> Dispatch(const OpExpr& op_expr, const TensorTuple& inputs) {return Dispatch<T>(op_expr, inputs, OpExprInterpContext(AttrMap{}));}template<typename T>static Maybe<T> Dispatch(const OpExpr& op_expr, const TensorTuple& inputs,const OpExprInterpContext& ctx);static Maybe<void> Dispatch(const OpExpr& op_expr, const TensorTuple& inputs,TensorTuple* outputs, const AttrMap& attrs) {return Dispatch(op_expr, inputs, outputs, OpExprInterpContext(attrs));}static Maybe<void> Dispatch(const OpExpr& op_expr, const TensorTuple& inputs,TensorTuple* outputs) {return Dispatch(op_expr, inputs, outputs, OpExprInterpContext(AttrMap{}));}static Maybe<void> Dispatch(const OpExpr& op_expr, const TensorTuple& inputs,TensorTuple* outputs, const OpExprInterpContext& ctx);
/* static */ Maybe<void> OpInterpUtil::Dispatch(const OpExpr& op_expr,const TensorTuple& inputs,TensorTuple* outputs,const OpExprInterpContext& ctx) {return JUST(GetInterpreter(inputs, ctx, op_expr))->Apply(op_expr, inputs, outputs, ctx);}
Maybe<AutogradInterpreter> GetInterpreter(const TensorTuple& inputs, const OpExprInterpContext& ctx,const OpExpr& op_expr) {static const auto& g_lazy_interpreter = BuildLazyInterpreter();static const auto& g_eager_consistent_interpreter = BuildEagerInterpreter(/*is_mirrored=*/false);static const auto& g_eager_mirrored_interpreter = BuildEagerInterpreter(/*is_mirrored=*/true);if (!LazyMode::is_enabled()) {if (inputs.empty()) {if (ctx.parallel_desc.has_value()) {JUST(ctx.nd_sbp);CHECK_OR_RETURN(!ctx.device.has_value());return g_eager_consistent_interpreter;} else {CHECK_OR_RETURN(!ctx.nd_sbp.has_value());return g_eager_mirrored_interpreter;}} else {if (inputs.at(0)->is_consistent()) {...return g_eager_consistent_interpreter;} else {...return g_eager_mirrored_interpreter;}}UNIMPLEMENTED_THEN_RETURN();}return g_lazy_interpreter;}
EagerMirroredInterpreter
EagerConsistentInterpreter
LazyInterpreter
std::shared_ptr<AutogradInterpreter> BuildEagerInterpreter(const bool& is_mirrored) {std::shared_ptr<OpExprInterpreter> internal;if (is_mirrored) {internal = std::make_shared<EagerMirroredInterpreter>();} else {internal = std::make_shared<EagerConsistentInterpreter>();}return std::make_shared<AutogradInterpreter>(internal);}std::shared_ptr<AutogradInterpreter> BuildLazyInterpreter() {auto internal = std::make_shared<LazyInterpreter>();return std::make_shared<AutogradInterpreter>(internal);}
class AutogradInterpreter {public:AutogradInterpreter() = delete;AutogradInterpreter(const std::shared_ptr<OpExprInterpreter>& internal) : internal_(internal) {}virtual ~AutogradInterpreter() = default;Maybe<void> Apply(const OpExpr& op_expr, const TensorTuple& inputs, TensorTuple* outputs,const AttrMap& attrs) const {return Apply(op_expr, inputs, outputs, OpExprInterpContext(attrs));}Maybe<void> Apply(const OpExpr& op_expr, const TensorTuple& inputs, TensorTuple* outputs) const {return Apply(op_expr, inputs, outputs, OpExprInterpContext(AttrMap{}));}Maybe<void> Apply(const OpExpr& op_expr, const TensorTuple& inputs, TensorTuple* outputs,const OpExprInterpContext& ctx) const;private:std::shared_ptr<OpExprInterpreter> internal_;};
Maybe<void> AutogradInterpreter::Apply(const OpExpr& op_expr, const TensorTuple& inputs,TensorTuple* outputs, const OpExprInterpContext& ctx) const {// 判断是否需要计算梯度,如果处于GradMode的作用域切改op注册时没有禁用梯度// 则requires_grad的值根据输入tensor的requires_grad属性判断// any of input tensors requires_grad==True,则表示需要计算梯度bool requires_grad = false;if (autograd::GradMode::is_enabled() && !JUST(op_expr.IsGradDisabled())) {requires_grad =std::any_of(inputs.begin(), inputs.end(),[](const std::shared_ptr<Tensor>& tensor) { return tensor->requires_grad(); });}// 这一坨逻辑比较丑陋,是因为近期支持了oneflow系统中支持了stride&&view机制// 而大部分op尚未注册stride推导、尚未支持non-contiguous的输入tensor// 所以需要在这对这部分op的输入进行强制转换,将其变为contiguous的// NOTE: if this op not support stride, then need to tensor->contiguous()#define HANDLE_NON_CONTIGUOUS_INPUT(tensor_tuple_ptr) \TensorTuple tmp_inputs; \if (!LazyMode::is_enabled() && !JUST(op_expr.SupportNonContiguous())) { \tmp_inputs.resize(inputs.size()); \for (size_t i = 0; i < inputs.size(); i++) { tmp_inputs[i] = inputs[i]->contiguous(); } \tensor_tuple_ptr = &tmp_inputs; \}const TensorTuple* inputs_ptr = &inputs;HANDLE_NON_CONTIGUOUS_INPUT(inputs_ptr);// 这里是进行实际Interpreter执行的主要过程{autograd::AutoGradMode mode(false);JUST(internal_->Apply(op_expr, *inputs_ptr, outputs, ctx));}// 这里主要是为了eager mode下,且requires_grad==True的op,// 插入反向节点(AddNode)用于autograd,该节点包含反向梯度计算的方法(backward_fn)// Lazy mode will construct backward compute graph in passes, so disable autograd if lazy mode.std::shared_ptr<OpExprGradClosure> grad_closure(nullptr);if (requires_grad && !LazyMode::is_enabled()) {grad_closure = JUST(op_expr.GetOrCreateOpGradClosure());auto backward_fn = std::make_shared<BackwardFunction>();backward_fn->body = [=](const TensorTuple& out_grads, TensorTuple* in_grads,bool create_graph) -> Maybe<void> {autograd::AutoGradMode mode(create_graph);JUST(grad_closure->Apply(out_grads, in_grads));return Maybe<void>::Ok();};backward_fn->status = [=]() { return grad_closure->state()->SavedTensors().size() > 0; };JUST(GetThreadLocalAutogradEngine()->AddNode(op_expr.op_type_name() + "_backward", backward_fn,*inputs_ptr, outputs));}// Update outputs autograd meta// Note: if requires_grad is True, we will create a new autograd meta for each output// in `AddBackwardFuncPtr` to support inplace operation, so the update should after// `AddBackwardFuncPtr`for (auto& output : *outputs) {output->set_is_leaf(inputs_ptr->size() == 0 || !requires_grad);...if (!output->requires_grad()) {JUST(output->set_requires_grad(requires_grad && IsSupportRequireGradDataType(output->dtype()->data_type())));}}// 捕获前向的inputs outputs,反向计算时可能用到if (requires_grad && !LazyMode::is_enabled()) {// Capture inputs and outputs after `AddBackwardFuncPtr` because of that grad function// node has been attached to them.JUST(grad_closure->Capture(*inputs_ptr, *outputs, ctx));}return Maybe<void>::Ok();}
// 这里是进行实际Interpreter执行的主要过程{autograd::AutoGradMode mode(false);JUST(internal_->Apply(op_expr, *inputs_ptr, outputs, ctx));}
Maybe<void> EagerInterpreter::Apply(const OpExpr& op_expr, const TensorTuple& inputs,TensorTuple* outputs, const OpExprInterpContext& ctx) const {(const auto* op = dynamic_castreturn ApplyImpl(*op, inputs, outputs, ctx); \}APPLY_IF(UserOp);APPLY_IF(VariableOp);APPLY_IF(CastToMirroredOp);APPLY_IF(CastFromMirroredOp);APPLY_IF(ConsistentToConsistentOp);APPLY_IF(CastToConsistentOp);APPLY_IF(CastFromConsistentOp);APPLY_IF(DistributeSplitOp);APPLY_IF(DistributeCloneOp);APPLY_IF(DistributeConcatOp);APPLY_IF(DistributeAddOp);APPLY_IF(FunctionOp);APPLY_IF(SelectTopNOp)OF_UNIMPLEMENTED() << "The type " << op_expr.op_type_name()<< " has not been supported in EagerInterpreter::Apply.";}
if (const auto* op = dynamic_cast<const UserOpExpr*>(&op_expr)) {return ApplyImpl(*op, inputs, outputs, ctx);}
Maybe<void> EagerMirroredInterpreter::ApplyImpl(const UserOpExpr& op_expr,const TensorTuple& inputs, TensorTuple* outputs,const OpExprInterpContext& ctx) const {return NaiveInterpret(op_expr, inputs, outputs, ctx);}
check input tensor的device是否一致
生成output tensor
为output tensor推导和检查shape/stride/dtype
构建op执行指令,并派发至vm
Maybe<void> NaiveInterpret(const UserOpExpr& user_op_expr, const TensorTuple& inputs,const Symbol<Device>& default_device, TensorTuple* outputs,const OpExprInterpContext& ctx) {const auto& attrs = ctx.attrs;std::shared_ptr<EagerBlobObjectList> input_eager_blob_objects =std::make_shared<EagerBlobObjectList>(inputs.size());// check devicesfor (int i = 0; i < inputs.size(); i++) {const auto& input_device = JUST(inputs.at(i)->device());if (i > 0) {CHECK_OR_RETURN(*default_device == *input_device)<< Error::RuntimeError()<< "Expected all tensors to be on the same device, but found at least two devices, "<< default_device->ToString() << " (positional 0) and " << input_device->ToString()<< " (positional " << i << ")!";}input_eager_blob_objects->at(i) = JUST(inputs.at(i)->eager_blob_object());}// make output tensorsstd::shared_ptr<EagerBlobObjectList> output_eager_blob_objects =std::make_shared<EagerBlobObjectList>(outputs->size());auto* output_tensor_metas = ThreadLocalDefaultOutputMutTensorMetas(outputs->size());for (int i = 0; i < outputs->size(); i++) {if (!outputs->at(i)) {const auto& tensor_impl = std::make_shared<EagerMirroredTensorImpl>();outputs->at(i) = std::make_shared<MirroredTensor>(tensor_impl);output_tensor_metas->at(i) = tensor_impl->mut_tensor_meta();} else {bool has_eager_blob_object = JUST(outputs->at(i)->has_eager_blob_object());CHECK_OR_RETURN(has_eager_blob_object);output_eager_blob_objects->at(i) = JUST(outputs->at(i)->eager_blob_object());}}Symbol<Stream> stream;bool need_check_mem_case = true;// Infer devices...// Infer shapes strides dtype...// 构建op执行指令,并派发至vmJUST(PhysicalRun([&](InstructionsBuilder* builder) -> Maybe<void> {return builder->LocalCallOpKernel(kernel, input_eager_blob_objects, output_eager_blob_objects,ctx, stream);}));return Maybe<void>::Ok();}
oneflow/core/eager/opkernel_instruction_type.cpp
static inline void OpKernelCompute(LocalCallOpKernelPhyInstrOperand* operand,DeviceCtx* device_ctx, user_op::OpKernelState* state,const user_op::OpKernelCache* cache) {auto* opkernel = operand->mut_opkernel();auto* compute_ctx =opkernel->UpdateComputeContext(operand->inputs().get(), operand->outputs().get(),operand->consistent_tensor_infer_result().get(), device_ctx);...operand->user_opkernel()->Compute(compute_ctx, state, cache);opkernel->UpdateComputeContext(nullptr, nullptr, nullptr, nullptr);}
operand->user_opkernel()->Compute(compute_ctx, state, cache);
oneflow/user/kernels/xxx_kernel.cpp
oneflow/user/kernels/xxx_kernel.cu
class UnaryPrimitiveKernel final : public user_op::OpKernel, public user_op::CudaGraphSupport {public:OF_DISALLOW_COPY_AND_MOVE(UnaryPrimitiveKernel);UnaryPrimitiveKernel() = default;~UnaryPrimitiveKernel() = default;using PrimitiveFactoryFuncType = std::function<std::unique_ptr<ep::primitive::ElementwiseUnary>(user_op::KernelComputeContext*)>;UnaryPrimitiveKernel(const std::string& output_name, const std::string& input_name,PrimitiveFactoryFuncType fn): output_name_(output_name),input_name_(input_name),primitive_factory_func_(std::move(fn)) {}private:using user_op::OpKernel::Compute;void Compute(user_op::KernelComputeContext* ctx) const override {auto primitive = primitive_factory_func_(ctx);CHECK(primitive);const user_op::Tensor* input_tensor = ctx->Tensor4ArgNameAndIndex(input_name_, 0);...const int64_t elem_cnt = input_shape.elem_cnt();if (elem_cnt != 0) {primitive->Launch(ctx->stream(), input_tensor->dptr(), output_tensor->mut_dptr(), elem_cnt);}}bool AlwaysComputeWhenAllOutputsEmpty() const override { return false; }std::string output_name_;std::string input_name_;PrimitiveFactoryFuncType primitive_factory_func_;};
template<UnaryOp unary_op, typename Src, typename Dst>class ElementwiseUnaryImpl : public ElementwiseUnary {public:OF_DISALLOW_COPY_AND_MOVE(ElementwiseUnaryImpl);ElementwiseUnaryImpl(Scalar attr0, Scalar attr1) : attr0(attr0), attr1(attr1) {}~ElementwiseUnaryImpl() override = default;void Launch(Stream* stream, const void* src_ptr, void* dst_ptr, size_t count) override {CpuStream* cpu_stream = stream->As<CpuStream>();Dst* dst = reinterpret_cast<Dst*>(dst_ptr);const Src* src = reinterpret_cast<const Src*>(src_ptr);auto functor = UnaryFunctor<DeviceType::kCPU, unary_op, Dst, Src>(attr0, attr1);cpu_stream->ParallelFor(0, count, [functor, src, dst](int64_t begin, int64_t end) {for (int64_t i = begin; i < end; i++) { dst[i] = functor(src[i]); }});}protected:Scalar attr0, attr1;};
template<DeviceType device, typename Dst, typename Src>struct UnaryFunctor<device, UnaryOp::kRelu, Dst, Src> {UnaryFunctor(Scalar attr0, Scalar attr1) {}OF_DEVICE_FUNC Dst operator()(Src src) const {const Src zero_val = static_cast<Src>(0.0);if (src <= zero_val) {return static_cast<Dst>(zero_val);} else {return static_cast<Dst>(src);}}};

原文作者:OneFlow;如有侵权,请联系管理员删除。