3月底整理了一个关于经典Paged Attention算法的ppt, 想起这个几年没写过的blog，把PPT改成一篇文章证明我还活着(-_-)。

vLLM 的 Paged Attention

开始前先说明一下，vLLM里的Paged Attention Kernel是有好几个不同的版本的，大概是下面这样子：

vLLM早期版本：

• Prefilling -> Flash Attention的flash_attn_varlen_func
• Dedocding -> 自己实现的Paged Attention
  • paged_attention_v1 : 用于比较短的sequence
  • paged_attention_v2 : 用于不想用v1的情况 :)

源码大概是这样的：

    # NOTE(woosuk): We use a simple heuristic to decide whether to use
    # PagedAttention V1 or V2. If the number of partitions is 1, we use
    # V1 to avoid the overhead of reduction. Also, if the number of
    # sequences or heads is large, we use V1 since there is enough work
    # to parallelize.
    # TODO(woosuk): Tune this heuristic.
    # For context len > 8192, use V2 kernel to avoid shared memory
    # shortage.
    use_v1 = (max_seq_len <= 8192 and (max_num_partitions == 1 or num_seqs * num_heads > 512))

vLLM 最新版本就已经全部转向Flash Attention， 用cutlass实现了。

NVIDIA GPU 基础

在深入 Paged Attention 的实现之前，我们需要了解 NVIDIA GPU的基本架构。（这里我们主要讲A100）

在做开发时，GPU 的 CUDA 程序包括 Grid -> Thread Block -> Threads三层架构。 这三层架构对应到GPU的硬件：GPU -> SM -> Cuda Core

在实际执行的时候，Threads会以每32个为一组执行，因此这里还多了一层：Thread Wrap，因此结构变成：

• CUDA程序：Grid -> Thread Block -> Thread Wrap -> Threads四层架构。
• GPU硬件：GPU -> SM(多次执行) -> SM(一次执行) -> Cuda Core，也是四层。

在 A100 GPU 上，我们有：

背景

多目标优化中有一个很常见的跷跷板问题，就是说在训练时，多个目标会相互影响，导致震荡—你降我升，我升你降。有时间还会出现Nan的结果，需要很仔细的调参测试+清洗数据才能训练出一个理想的模型。

针对这种问题，自然就有了一些尝试，比如从帕累托最优的角度寻找优化方向（阿里PEA），修改模型结构使Shared部分存储更泛化的信息（腾讯PLE）。不过这两个写的人都挺多了，就写一下Google Small Towers的这篇文章吧。

主要问题讨论

文章首先讨论了两个问题：

1. Over-parameterization对多任务模型的适用性

我们都知道over-parameterization对单任务模型是有价值的，那边对多任务模型是否成立？

这里以将多个目标的线性组合作为优化目标的例子，认为over-parameterization能够帮助处理各任务优化目标之间的冲突问题（既减少跷跷板问题的出现）。

2. 大型模型和小型模型的多目标学习表现对比

通过实验对比了大型模型和小型模型进行多目标学习中的不同表现。

实验中，不论是增加任务相关结构的复杂度，还是增加任务共享结构的复杂度，Pareto frontier都会呈现先变好在变差的趋势。

因此，文章认为over-parameterization并不利于多目标学习中的共享性，进而伤害了多目标学习中的泛化能力。因此，在多目标学习中，模型大小实质上是对模型有效性和泛化能力的一种平衡。

好久没写博客了，今天写一点推荐系统周边设施的东西。

特征管理

特征商店会存储特征元数据，比如特征的计算逻辑、血缘关系、数据类型。 一般来说，这些元数据用于管理特征的生命周期、计算任务和使用方式。

离线训练数据生成

为了保证线上线下数据的一致性，推荐系统的训练数据通常有两个数据流Join得到：

• 在Ranking中即实时打点：数据流以traceId为Key，排序时特征为Value。
• 客户端日志：记录了traceId和事件类型(曝光、点击、分享等）

由于客户端日志必然晚于服务端日志，因此两个数据流Join时需要一定的窗口。

训练数据扩展

但是作为调参工程师，我们必然会遇到需要的特征没有记录在实时打点中，导致训练时缺少相关数据的情况，这个时候，就需要想办法来处理这个问题。

按照Uber的方法，我们可以把特征分为三类：

1. 离线特征
2. 实时特征
3. RPC特征

离线特征

对于离线特征：我们可以使用Spark读取数据仓库中的历史数据，以天为单位进行生成历史数据，然后放在一个分区的Hive表中。

实时特征

对于实时特征：基于kappa的思想，我们可以在Flink中编写实时特征计算逻辑，然后启动重跑一段时间以前的历史数据，并记录这个过程中特征的每一次变化（有点类似数据库中的WAL日志流），将其输出到Kafka中去，这样我们也就有一个特征在历史时间段中的值。(这里我们最好有一个服务化的Flink平台，来进行任务的添加、删除、修改等工作)

NPE(NullPointerException)是一个很烦人的问题，这里简单列举了一些语言中对NPE的处理。

1. 通过语法标记进行检查

Kotlin

Kotlin要求可以为null的变量必需在定义时声明，同时在读取该类型变量属性时必须进行空值判断。例：String 和 String?

var a: String = "abc"
a = null // compilation error, a can not be null

var b: String? = "abc"
b = null // ok

val l = b.length // compiler error: variable 'b' can be null
val l = if (b != null) b.length else -1 // ok

Jetbrains annotations for Java

IntelliJ IDEA提供了一些工具，比如可以对@NotNull的参数进行检查，当出现null赋值时在IDE中会给出提示。

import org.jetbrains.annotations.NotNull;
import java.util.ArrayList;

public class Test{
    public void foo(@NotNull Object param){
        int i = param.hashCode();
    }
    
    public void test(){
        foo(null); // warn in IntelliJ IDEA
    }
}

（类似的，FindBugs也提供了@Nonnull注释，用于检查）

Lombok for Java

Lombok通过在编译时改写字节码对原始代码进行优化，其中的@NonNull，会自动插入运行时检查代码，发现错误抛出异常。

public NonNullExample(@NonNull Person person) {
  super("Hello");
  this.name = person.getName();
}

等价于

public NonNullExample(Person person) {
  if (person == null) {
    throw new NullPointerException("person is marked @NonNull but is null");
  }
  super("Hello");
  this.name = person.getName();
}

2. 用更好的错误处理代替null

空值通常都是由错误导致的无法赋值，因此更好的错误处理也是NPE的一种应对。

Rust：基于Result错误处理

Rust通过Result类型提供了强大的错误处理机制。

基于Monad处理错误

Scala等FP语言基于Monad(Option, Either, Try…)提供了错误处理，其中Optional是最基础的一种。在Option中，定义了专门的None来表示计算失败。这样，在得不到结果时，就会得到None，因此在后续的使用中可以使用isDefined先判断是否有值，再进行处理：

HttpServerModule的请求主要由HttpServer中的HttpServerTransport （默认为NettyHttpServerTransport）类处理。

NettyHttpServerTransport基于netty框架，负责监听并建立连接，信息的处理由内部类HttpChannelPipelineFactory 完成。

每当产生一个连接时，都会发出一个ChannelEvent，该Event由一系列的ChannelHandler进行处理。

为了方便组织，这些ChannelHandler被放在一条“流(pipeline)”里，一个ChannelEvent并不会主动的”流”经所有的Handler，而是由上一个Handler显式的调用ChannelPipeline.sendUp(Down)stream产生，并交给下一个Handler处理。

换句话说，每个Handler接收到一个ChannelEvent，并处理结束后，如果需要继续处理，那么它需要调用sendUp(Down)stream新发起一个事件。如果它不再发起事件，那么处理就到此结束，即使它后面仍然有Handler没有执行。这个机制可以保证最大的灵活性，当然对Handler的先后顺序也有了更严格的要求。

Tornado是一款轻量级的Web服务器，同时又是一个开发框架。采用单线程非阻塞I/O模型(epoll)，主要是为了应对高并发 访问量而被开发出来，尤其适用于comet应用。

Tornado服务器3大核心模块:

(1) IOLoop

Tornado为了实现高并发和高性能，使用了一个IOLoop来处理socket的读写事件，IOLoop基于epoll，可以高效的响应网络事件。这是Tornado高效的保证。

tornado.ioloop.IOLoop.instance().start()

IOLoop使用了单例模式，处理所有IO事件，

实现为EPollIOLoop->PollIOLoop->IOLoop->Configurable

IOLoop中有四个重要的数据集： _events 和 _handlers 保存I/O事件和对应的处理器， _callbacks 和 _timeouts 保存（超时）回调。

关键函数：

def initialize(self, impl, time_func=None):
    super(PollIOLoop, self).initialize()
    self._impl = impl
    if hasattr(self._impl, 'fileno'):
        set_close_exec(self._impl.fileno())
    self.time_func = time_func or time.time
    #handlers 是一个函数集字典
    self._handlers = {}
    self._events = {}
    #回调函数集合
    self._callbacks = []
    self._callback_lock = threading.Lock()
    self._timeouts = []
    self._cancellations = 0
    self._running = False
    self._stopped = False
    self._closing = False
    self._thread_ident = None
    self._blocking_signal_threshold = None
    self._timeout_counter = itertools.count()

    # Create a pipe that we send bogus data to when we want to wake
    # the I/O loop when it is idle
    self._waker = Waker()
    self.add_handler(self._waker.fileno(),
                     lambda fd, events: self._waker.consume(),
                     self.READ)

其中，waker是一个发伪数据用的类，在需要时，我们可以用它唤醒空闲的I/O Loop。当我们调用add_callback时，为了让回调函数运行，可能会需要使用它发送一个伪数据。