标签: dataflow

什么是数据流编程语言？为什么要用它？它有什么好处吗？

低延迟数据访问是什么意思？

我实际上对这个术语的定义感到困惑."LATENCY"

任何人都可以详细说明"延迟"一词.

我真的看不出它们之间的区别.它们都是关于流经指令的数据和输入数据中的变化传播.我读过这本书(由Matt Carcki撰写),它清楚地表明两者都是一样的.另一方面,维基百科建立了反应式编程作为数据流编程的一种形式,这种堆栈溢出的答案也是如此.

那么,Reactive编程和Dataflow编程之间的概念差异是什么？

我目前正在为连续环境中的编程开发一种新语言(将其与电气工程相比较),并且我对某种语言结构有一些想法.

让我通过解释然后按定义解释这个特征:

x = a U b;

x变量在哪里,a而且b是其他变量(或静态值).这就像是a和之间的结合b; 没有重复,也没有具体的订单.

with(x) {
    // regular 'with' usage; using the global interpretation of "x"
    x = 5;
    // effectively will do:
    // x = a U b U 5;
    // a = 5;
    // b = 5;
    // Thus, when "a" or "b" changes, "x" is still equal to "5".
}
with(x = a) {
    // this code block is executed when the "x" …

我不是在寻找神经网络库,因为我正在创建新的网络.为此我需要一个好的"数据流"语言.

当然,你可以用C,C++,Java和co做到这一点.但是从头开始处理多线程等将是一场噩梦.

在另一个极端,像Oz或Erlang这样的语言看起来更适应,但它们没有很多库,而且它们更难掌握(它们很容易使用它们,但创建完整的软件是否可以？).

你会建议什么？

交叉发布到http://social.msdn.microsoft.com/Forums/en-US/tpldataflow/thread/89b3f71d-3777-4fad-9c11-50d8dc81a4a9

我知道......我并没有真正使用TplDataflow来发挥它的最大潜力.ATM我只是BufferBlock用作消息传递的安全队列,其中生产者和消费者以不同的速率运行.我看到一些奇怪的行为,让我难以理解如何继续.

private BufferBlock<object> messageQueue = new BufferBlock<object>();

public void Send(object message)
{
    var accepted=messageQueue.Post(message);
    logger.Info("Send message was called qlen = {0} accepted={1}",
    messageQueue.Count,accepted);
}

public async Task<object> GetMessageAsync()
{
    try
    {
        var m = await messageQueue.ReceiveAsync(TimeSpan.FromSeconds(30));
        //despite messageQueue.Count>0 next line 
        //occasionally does not execute
        logger.Info("message received");
        //.......
    }
    catch(TimeoutException)
    {
        //do something
    }
}

在上面的代码中(它是2000行分布式解决方案的一部分),Send每100ms左右定期调用一次.这意味着一个项目被Post编到messageQueue在约10次.这已经过验证.但是,偶尔看起来ReceiveAsync在超时内没有完成(即Post没有导致ReceiveAsync完成)并且TimeoutException在30秒后被提升.在这一点上,messageQueue.Count是数百.这是出乎意料的.这个问题已经在较慢的发布率(1个帖子/秒)中观察到,并且通常在1000个项目通过之前发生BufferBlock.

因此,要解决此问题,我使用以下代码,它可以工作,但偶尔会在接收时导致1秒延迟(由于上面发生的错误)

    public async …

我刚刚遇到了提议的Boost :: Dataflow库.这似乎是一个有趣的方法,我想知道是否有其他这样的替代框架的C++,以及是否有任何相关的设计模式.我没有排除Boost :: Dataflow,我只是在研究任何可用的替代品,这样我就能更好地理解域和我的选择(或者如果有必要,可以自己动手).

我的问题如下:我有一些python类具有从其他属性派生的属性; 这些应该在计算后缓存,并且每次更改基本属性时缓存的结果都应该无效.

我可以手动完成,但如果属性数量增加,似乎很难维护.所以我希望在我的对象中有类似Makefile规则的东西来自动跟踪需要重新计算的内容.

所需的语法和行为应该是这样的:

# this does dirty magic, like generating the reverse dependency graph,
# and preparing the setters that invalidate the cached values
@dataflow_class
class Test(object):

    def calc_a(self):
        return self.b + self.c

    def calc_c(self):
        return self.d * 2

    a = managed_property(calculate=calc_a, depends_on=('b', 'c'))
    b = managed_property(default=0)
    c = managed_property(calculate=calc_c, depends_on=('d',))
    d = managed_property(default=0)


t = Test()

print t.a
# a has not been initialized, so it calls calc_a
# gets b value
# c has not …

我特别想要在一个或另一个中编写一些信号处理算法,或者两者兼而有之.

表现并不是一个大问题,表达意图的清晰度更为重要.

我想要实现以下'Blocks'并组成它们:

• 滤波器(FIR和IIR)
• 相位检测器
• 集成
• 搅拌机
• 函数发生器
• PLL(使用上面的构建块)

我认为Rx可以被认为是'Linq-to-streams',而TPL是对并发的抽象.我还得到Rx在内部使用TPL来管理其异步位,并且TPL数据流为TPL增加了可组合性.

所以两者都是异步的,都是可组合的,都是相当高的水平(Rx更多).通常和上面的信号处理项目应该在哪里使用？

我有一个执行一些不稳定计算的 luigi 任务。想想有时不收敛的优化过程。

import luigi

MyOptimizer(luigi.Task):
    input_param: luigi.Parameter()
    output_filename = luigi.Parameter(default='result.json')

    def run(self):
        optimize_something(self.input_param, self.output().path)

    def output(self):
        return luigi.LocalTarget(self.output_filename)

现在我想构建一个包装器任务，该任务将使用不同的输入参数多次运行此优化器，并获取收敛的第一次运行的输出。

我现在实现它的方式是不使用，MyOptimizer因为如果它失败，luigi 会认为包装器任务也失败了，但我可以接受一些MyOptimizer失败的实例。

MyWrapper(luigi.Task):
    input_params_list = luigi.ListParameter()
    output_filename = luigi.Parameter(default='result.json')

    def run(self):
        for input_param in self.input_params_list:
            try:
                optimize_something(self.input_param, self.output().path)
                print(f"Optimizer succeeded with input {input_param}")
                break
            except Exception as e:
                print(f"Optimizer failed with input {input_param}. Trying again...")

    def output(self):
        return luigi.LocalTarget(self.output_filename)

问题在于，通过这种方式，任务不会并行化。此外，您可以想象MyOptimizer和optimize_something参与由 luigi 处理的数据管道的复杂任务，这在我的代码中造成了相当多的混乱。

我将不胜感激有关如何以类似路易吉的方式进行这项工作的任何见解和想法:)