DFS通用解法

May 8th, 2015 by JasonLe's Tech 1,751 views

最近在刷一些算法题,发现DFS在单链表,二叉树,图,集合的解题比较多,具有一定的通用规律,现在讲通用方法记录下。拿二叉树举例,比如我们需要从根走到叶子节点才能得到一个解,这种类型非常适合是用DFS,再以二维数组举例,我们可以将二维数组当成一个图,进行搜索,在搜索的同事满足一定的匹配等。

一般情况下Wide-FS只要求有一个解,而且需要将整个中间状态存储到内存中,而DFS只存储一条路径,非常时候解决一些问题。

在DFS中我们需要一个收敛条件,也就是合法解。这时我们就需要把这个中间状态保存到最后的结果中。为了加快深搜,我们可以剪枝,常用方式使用状态数组表示,提前return,可以大大加快递归速度。

通用dfs模板:

/**
* dfs 模板.
* @param[in] input 输入数据指针
* @param[out] path 当前路径,也是中间结果,可以是一维数组
* @param[out] result 存放最终结果,二维数组
* @param[inout] cur or gap 标记当前位置或距离目标的距离,或者可以
* 是start end等标记
* @return 路径长度,如果是求路径本身,则不需要返回长度
* 可以返回bool等,依照题目要求来实现。
*/
void dfs(type &input, type &path, type &result, int cur or gap) {
              if (数据非法) return 0; // 终止条件
              if (cur == input.size()) { // 收敛条件
                  // if (gap == 0) {
                        将path 放入result
              }
              if (可以剪枝) return;
              for(...) { // 执行所有可能的扩展动作
                     执行动作,修改path
                     dfs(input, step + 1 or gap--, result);
                     恢复path
              }
}

这里我举一个例子:列举所有set可能的子集合,比如S=[1,2,3],那么结果是[[3],[2],[1],[1,2,3],[1,3],[2,3],[1,2],[]]
解决这个问题,需要首先按照上面的这种模板构建,首先是这个dfs的input 也就是这个S,中间路径path与S类型相同。结果应该是一个二维数组,也就是vector< vector > result,最后我们需要一个step作为收敛条件。

void dfs(const vector<int> &S, vector<int> &path, vector<vector<int> > &result,int step) {
     if (step == S.size()) {//到达S.size()收敛
           result.push_back(path);
           return;
     }
     //这里没有剪枝
     // 不选S[step]
     subsets(S, path, step + 1, result);
     // 选S[step]
     path.push_back(S[step]);
     subsets(S, path, step + 1, result);
     path.pop_back();
}
 void dfs(vector<int>& nums,vector<int> &path,vector<vector<int>> &result,
                     vector<int>::iterator start){
 
     result.push_back(path);
 
     for(auto i = start;i<nums.end();i++)
     {
          path.push_back(*i);
          dfs(nums,path,result,i+1);
          path.pop_back(); 
     }
 }

深度搜索比较难以理解,层层递归会让我迷失,不过进行断点认真跟踪是可行的。最后跟踪断点结果是:

[]
3
2
2,3
1,
1,3
1,2
1,2,3

还有很多场景,比如二维数组寻路,都会用到上下左右的移动,还要使用flag来标示,具体查看
https://leetcode.com/problems/number-of-islands/
https://leetcode.com/problems/word-search/

物理内存管理:伙伴系统释放页框

April 24th, 2015 by JasonLe's Tech 1,861 views

相对与页框慢速分配快速分配而言。页框释放函数非常简单,主要的函数就是__free_pages()和__free_page()。很明显,类似于之前page分配的上层函数,也是层层wrapper。这里__free_page()是__free_pages()的包装。

#define __free_page(page) __free_pages((page), 0)

void __free_pages(struct page *page, unsigned int order)
{
        if (put_page_testzero(page)) {
                if (order == 0)
                        free_hot_cold_page(page, 0);
                else
                        __free_pages_ok(page, order);
        }
}

通过代码我们发现当要释放的页order==0,直接使用per-cpu缓存释放,大于0使用__free_pages_ok(page, order)释放。

这里kernel使用free_hot_cold_page() 方式释放order为0的页,这个函数只将不可移动页,可回收页和可移动页放入每个CPU页框高速缓存中,如果迁移类型不再这个范围,那么这个页要回收到伙伴系统中!这里的策略是:热页加入表头,冷页加入表尾。这个部分不在本文中叙述,查看代码便知。

来看__free_pages_ok()函数,这个函数也是一个wrapper函数,主要用来检查参数是否正确,做一些释放page前的准备工作。当一切就绪后就会调用free_one_page()函数。大致调用关系是:

static void __free_pages_ok(struct page *page, unsigned int order)
{
....
    free_one_page(page_zone(page), page, pfn, order, migratetype);
....
}
static void free_one_page(struct zone *zone,
                struct page *page, unsigned long pfn,
                unsigned int order,
                int migratetype)
{
    unsigned long nr_scanned;
    spin_lock(&zone->lock);
...
    __free_one_page(page, pfn, zone, order, migratetype);
    spin_unlock(&zone->lock);
}

在调用__free_one_page()之前,kernel换回更新当前zone下面page状态:更新当前内存管理区的空闲页面数,也就是更新zone下面的vm_stat数组,这个数组用于统计当前内存信息。

在核心函数__free_one_page()中,这个函数主要完成page的回收,完成页的合并,首先要获取要释放page的page index

static inline void __free_one_page(struct page *page,
        unsigned long pfn,
        struct zone *zone, unsigned int order,
        int migratetype)
{
    unsigned long page_idx;
    unsigned long combined_idx;
    unsigned long uninitialized_var(buddy_idx);
    struct page *buddy;
    int max_order = MAX_ORDER;

...
    page_idx = pfn & ((1 << max_order) - 1);
...
    while (order < max_order - 1) {
        buddy_idx = __find_buddy_index(page_idx, order);
        buddy = page + (buddy_idx - page_idx);
        if (!page_is_buddy(page, buddy, order))
            break;
        if (page_is_guard(buddy)) {
            clear_page_guard_flag(buddy);
            set_page_private(buddy, 0);
            if (!is_migrate_isolate(migratetype)) {
                __mod_zone_freepage_state(zone, 1 << order,
                              migratetype);
            }
        } else {
            list_del(&buddy->lru);
            zone->free_area[order].nr_free--;
            rmv_page_order(buddy);
        }
        combined_idx = buddy_idx & page_idx;
        page = page + (combined_idx - page_idx);
        page_idx = combined_idx;
        order++;
    }
    set_page_order(page, order);

    if ((order < MAX_ORDER-2) && pfn_valid_within(page_to_pfn(buddy))) {
        struct page *higher_page, *higher_buddy;
        combined_idx = buddy_idx &amp; page_idx;
        higher_page = page + (combined_idx - page_idx);
        buddy_idx = __find_buddy_index(combined_idx, order + 1);
        higher_buddy = higher_page + (buddy_idx - combined_idx);
        if (page_is_buddy(higher_page, higher_buddy, order + 1)) {
            list_add_tail(&page->lru,
                &zone->free_area[order].free_list[migratetype]);
            goto out;
        }
    list_add(&page->lru, &zone->free_area[order].free_list[migratetype]);
out:
    zone->free_area[order].nr_free++;
}

然遍历当前order到max-order所有阶,使用__page_find_buddy()找到当前page_idx的伙伴buddy_idx,然后根据idx偏移量,找到目标buddy。后将这个buddy从list删除,更新nr_free数量。删除buddy标记。最后使用 buddy_idx & page_idx 获取合并后的combined_idx。由于page永远都指向要释放页框块的首页框描述符,所以讲这个combined_idx赋值给page_idx。最后将order+1,然后通过set_page_order()设置这个page的各种信息。

在最新的源码里面,还会判断是否这个page是否是最大的页,如果找到则继续合并,并将合并后page放到list tail中,最后将当前order下的nr_order++。

最后我们来看__find_buddy_index()函数

static inline unsigned long
__find_buddy_index(unsigned long page_idx, unsigned int order)
{
      return page_idx ^ (1 << order);
}

这个函数用来查找buddy的index,这个函数非常简单。

综上所述,所以整个页框分配的核心就是那个while循环,我们可以把页框释放当成页框分裂的逆过程,也就是回收order大于0的页:
举例:

page_idx = 10 -> buddy_idx = 11 -> combined_idx = 10
page_idx = 10 -> buddy_idx = 8 -> combined_idx = 8
page_idx = 8 -> buddy_idx = 12 -> combined_idx = 8
page_idx = 8 -> buddy_idx = 0 -> combined_idx = 0
->无法继续进行->结束

这就是那些位操作的实际含义,page_idx以二倍速率寻找buddy page然后合并,至此伙伴系统回收页框完毕。

 

参考:

http://blog.csdn.net/vanbreaker/article/details/7624628

内存条物理结构分析

April 21st, 2015 by JasonLe's Tech 1,843 views

Update 2015-07-04

我们经常接触物理内存条,如下有一根DDR的内存条,我们可以看到这个内存条上面有8个黑色的内存颗粒,在高端服务器上面通常会带有ECC校验,所以会存在9个黑色的内存颗粒,其中一个的内存颗粒是专门做ECC校验的。

20150421160137

从概念的层次结构上面分为:Channel > DIMM > Rank > Chip > Bank > Row/Column

我们可以把DIMM作为一个内存条实体,我们知道一个内存条会有两个面,高端的内存条,两个面都有内存颗粒。所以我们把每个面叫做一个Rank,也就是说一个内存条会存在Rank0和Rank1。

拿rank0举例,上面有8个黑色颗粒,我们把每个黑色颗粒叫做chip。再向微观走,就是一个chip里面会有8个bank。每个bank就是数据存储的实体,这些bank就相当于一个二维矩阵,只要声明了column和row就可以从每个bank中取出8bit的数据。

我们之前会经常说双通道,说白了就是一个DIMM就是一个通道,两个DIMM组成双通道,分别由两个Memory Controller控制。

20150421162420

我们可以看到两个DIMM0 DIMM0组成双通道,两个DIMM1 DIMM1组成双通道。

下面先来解释memory controllers如何从rank中取数据,上面说的都是物理结构,下面说内存的逻辑结构。因为每个rank下面会有很多chip,而每个chip又包括bank0、bank1、bank2等,在memory controllers看来每次发数据,都会同时发送给所有chip下的某个bank,并声明row和col。

以从bank0为例:

20150421162433

每个chip的bank0 的同一地点(row=i col=j)都会被读出8bit,那么8个chip就会同时读出64bit,然后由memory controllers传送给cpu,也就是8byte。

在memory controllers看来,每个bank存在于每个chip中,如上图所示,可以把每个chip里面的小bank连成一行,b看作成一个大的bank。然后从大的bank中读取数据。

每个bank有一个row bufffer(这个row buffer存在于bank中,而不是在memory controller,row buffer 是以bank和row作为参数的,而读到memory controller时,又加入了col参数,也就是row buffer数据量是整整一行数据,读到memory controllers中仅仅是很小的一部分数据,而row buffer是整个程序局部性的关键!),作为一个bank page,所有bank共享地址、数据总线,但是每个channel有他们自己的地址、数据总线。正因为有buffer,所以每次bank都会预读64bit的数据。

上面看到的是分解的操作,事实上,为了加快memory的读写,体系结构中引入了流水线,也就意味着memory controllers可以同时读64byte,也就是8次这样的操作。写入到buffer中,这就是局部性原理。如果我们程序猿不尊重这个规则,也就迫使bank的buffer每次取值都必须清空当前的缓冲区,重新读数据,降低数据的访问速度。

 

 

设计弊端:

那么内存在多核平台的表现就取决于数据位于哪个bank之中,在给定的时间kernel如何访问多核之间共享的bank,最好的一种情况是每个core都只访问自己的bank,互不干扰。最坏的一种情况就是所有的core都访问同一个bank,只能同时有一个core访问该bank,其他core必须等待这个bank的,这样的话造成memory access的delay,考虑一种情况:如果多个进程在多个bank中都有自己的数据,controllers不得不每次清空row buffer,造成性能损失。而且使得时间分析变得不确定!

由于memory controllers内侧对于bank的操作对外透明,row col bank这些指令信息属于内存地址(memory controllers将physics address翻译成内存地址)。研究memory controllers向bank发送读信息的编码格式,我们会发现row与col位之间会有bank 位的介入,也就意味着对于bank的访问会分割到几个bank同时进行。

如果我们想把进程在内存中的数据限制在某个bank中,就要测试这种内存地址格式,目前可以palloc自带的工具进行测试,测试bank位到底存在于内存地址的哪几位。

目前系统都是多核,而且kernel将memory视为一个整体。不会区分分配的资源来自于哪个bank,所以数据分配的确定位置是不可预期的。而且目前memory controllers被配置为分割的bank来提高bank访问的并行度(一个程序分配的内存在每个bank中都存在)。但是这个导致一个问题:肯定有好几个进程数据存在于当前bank。当bank中出现multiple error时,我们无法准确定位这个错误来自于哪个进程!

 

[1] http://en.wikipedia.org/wiki/Memory_bank

[2] http://arxiv.org/pdf/1407.7448.pdf

priority_queue与heap的使用

April 20th, 2015 by JasonLe's Tech 1,674 views

1.priority_queue

priority_queue是一个优先队列,下面是他的声明,我们平时可以直接使用下面的方式声明一个优先队列。

priority_queue<int> pq

优先队列内部是一个heap的实现,也就是说默认push到priority_queue中的数据,当我们pop出来的时候,默认是优先级最高的,(数字大的优先级高,数字小的优先级低),这个数据结构默认使用vector作为容器,cmp函数默认使用less作为比较函数。

下面的是一个完整的priority_queue的声明

std::priority_queue
template <class T, class Container = vector<T>,
class Compare = less < typename Container::value_type> > class priority_queue;

 

priority_queue<Type, Container, Functional>
其中Type 为数据类型, Container 为保存数据的容器,Functional 为元素比较方式。Container 必须是用数组实现的容器,比如 vector, deque 但不能用 list。STL里面默认用的是 vector. 比较方式默认用 operator< , 所以如果把后面俩个参数缺省的话,优先队列就是大顶堆,队头元素最大。

我们使用的时候和平常queue的方式没有什么太大的却别,最大的区别在于这个cmp应该如何自定义。我们知道cmp是一个函数指针,所以我们可以有两种方式重载cmp函数。

 

struct cmp
{
    bool operator () (int &a, int &b)
    {
        return a > b ;              // 从小到大排序,值 小的 优先级别高
    }
}; 

priority_queue<int,vector<int>,cmp> q;

 

方式1:

struct Time {
    int h;
    int m;
    int s;
};

class CompareTime {
    public:
    bool operator()(Time& t1, Time& t2) // Returns true if t1 is earlier than t2
    {
       if (t1.h < t2.h) return true;
       if (t1.h == t2.h && t1.m < t2.m) return true;
       if (t1.h == t2.h && t1.m == t2.m && t1.s < t2.s) return true;
       return false;
    }
}

这里我们必须保证重载的()函数返回值是bool,上面的重载函数核心就是当t1<t2时候,返回tree,所以得到的也就是从大到小的排列,也是这个数据结构默认的,如果我们想重新实现这个数据结构,改为从小到大排列,那么可以使用下面的方式

方式2:

class CompareTime {
public:
    bool operator()(Time& t1, Time& t2) // t2 has highest prio than t1 if t2 is earlier than t1
    {
       if (t1.h > t2.h) return true;
       if (t2.h == t1.h && t2.m < t1.m) return true;
       if (t2.h == t1.h && t2.m == t1.m && t2.s < t1.s) return true;
       return false;
    }
};

保证第一个大于第二个返回true即可。
上面我们看到在一个class类里面重载()函数,我们也可以在要使用的类里面,使用struct{}方式。

class Solution {
public:
.....
private:
struct cmp {
        bool operator()(ListNode* node1, ListNode* node2) {
            return node1->val > node2->val;
        }
    };
};

在C/C++中,我们可以等同class与struct相似。

2.heap

heap 主要分为push_heappop_heapsort_heapreverse四个函数,我们使用这四个函数使得vector中数据按照heap来排列。

make_heap的两种形式:

template <class RandomAccessIterator>
  void make_heap (RandomAccessIterator first, RandomAccessIterator last);
template <class RandomAccessIterator, class Compare>
  void make_heap (RandomAccessIterator first, RandomAccessIterator last,
                  Compare comp );

同样有一个comp函数可以指定以排列顺序,所以priority_queue是基于heap的方式来实现的。

示例代码:

#include <iostream>
#include <algorithm>
#include <vector>

using namespace std;

class priority_queue
{
    private:
        vector<int> data;
    public:
        void push( int t ){
            data.push_back(t);
            push_heap( data.begin(), data.end());
        }
        void pop(){
            pop_heap( data.begin(), data.end() );
            data.pop_back();
        }
        int top() { return data.front(); }
        int size() { return data.size(); }
        bool empty() { return data.empty(); }
}; 

int main()
{
    priority_queue test;
    test.push( 3 );
    test.push( 5 );
    test.push( 2 );
    test.push( 4 );

    while( !test.empty() ){
        cout << test.top() << endl;
        test.pop(); }
    return 0;
}

 

参考:

[1] http://comsci.liu.edu/~jrodriguez/cs631sp08/c++priorityqueue.html

[2] http://www.cplusplus.com/reference/queue/priority_queue/

[3] http://stackoverflow.com/questions/23529815/how-to-use-stdmake-heap

[4] http://www.cppblog.com/mzty/archive/2005/12/15/1770.html

物理内存管理:伙伴系统page分裂函数分析

April 14th, 2015 by JasonLe's Tech 1,841 views

快速分配函数中,大部分代码都是一些具体策略的实现,包括watermark,NUMA,公平分配ALLOC_FAIR等。真正与buddy system有关的就是buffered_rmqueue()函数。

我们看到当进入到这个函数,会先对order进行一个判断,如果order==0,则kernel不会从伙伴系统分配,而是从per-cpu缓存加速请求的处理。如果缓存为空就要调用rmqueue_bulk()函数填充缓存,道理还是从伙伴系统中移出一页,添加到缓存。

当传入的order>0时,则调用_rmqueue()从伙伴系统中选择适合的内存块,有可能会将大的内存块分裂成为小的内存块,用来满足分配请求。

static inline
struct page *buffered_rmqueue(struct zone *preferred_zone,
            struct zone *zone, unsigned int order,
            gfp_t gfp_flags, int migratetype)
{
    unsigned long flags;
    struct page *page;
    bool cold = ((gfp_flags & __GFP_COLD) != 0);

again:
    if (likely(order == 0)) {
...
    } else {
        if (unlikely(gfp_flags & __GFP_NOFAIL)) {
            WARN_ON_ONCE(order > 1);
        }
        spin_lock_irqsave(&zone->lock, flags);
        page = __rmqueue(zone, order, migratetype);
        spin_unlock(&zone->lock);
        if (!page)
            goto failed;
        __mod_zone_freepage_state(zone, -(1 << order),
                      get_freepage_migratetype(page));
    }
...
}
...
static struct page *__rmqueue(struct zone *zone, unsigned int order,
                        int migratetype)
{
    struct page *page;

retry_reserve:
    page = __rmqueue_smallest(zone, order, migratetype);

    if (unlikely(!page) && migratetype != MIGRATE_RESERVE) {
        page = __rmqueue_fallback(zone, order, migratetype);

        if (!page) {
            migratetype = MIGRATE_RESERVE;
            goto retry_reserve;
        }
    }

    trace_mm_page_alloc_zone_locked(page, order, migratetype);
    return page;
}

上面说的buffered_rmqueue()是进入伙伴系统的前置函数,而__rmqueue是进入伙伴系统的最后一个包装函数,通过这个函数,__rmqueue_smallest()会扫描当前zone下面的空闲区域。

如果不能通过这种方式分配出空闲页,那么系统会调用__rmqueue_fallback()来遍历不同的迁移类型,试图找出不同的迁移类型中空闲的page。这里我们不会仔细分析,但是我们看一下__rmqueue_fallback()的遍历头就能发现端倪。kernel会从start_migratetype迁移类型考虑备用列表的不同迁移类型,具体可以查看http://lxr.free-electrons.com/source/mm/page_alloc.c#L1036

这里找到一篇叙述迁移页的博客《通过迁移类型分组来实现反碎片》,讲的很清楚!

static inline struct page *
__rmqueue_fallback(struct zone *zone, unsigned int order, int start_migratetype)
{
    struct free_area *area;
    unsigned int current_order;
    struct page *page;
    int migratetype, new_type, i;

    /* Find the largest possible block of pages in the other list */
    for (current_order = MAX_ORDER-1;
                current_order >= order && current_order <= MAX_ORDER-1;
                --current_order) {
        for (i = 0;; i++) {
            migratetype = fallbacks[start_migratetype][i];

...

__rmqueue_smallest()函数传入的参数有order,kernel会遍历当前zone下面从指定order到MAX_ORDER的伙伴系统,这个时候迁移类型是指定的,如图:
buddy

static inline
struct page *__rmqueue_smallest(struct zone *zone, unsigned int order,
                        int migratetype)
{
    unsigned int current_order;
    struct free_area *area;
    struct page *page;

    /* Find a page of the appropriate size in the preferred list */
    for (current_order = order; current_order < MAX_ORDER; ++current_order) {
        area = &(zone->free_area[current_order]);
        if (list_empty(&area->free_list[migratetype]))
            continue;

        page = list_entry(area->free_list[migratetype].next,
                            struct page, lru);
        list_del(&page->lru);
        rmv_page_order(page);
        area->nr_free--;
        expand(zone, page, order, current_order, area, migratetype);
        set_freepage_migratetype(page, migratetype);
        return page;
    }

    return NULL;
}

我们看到kernel会提取出当前zone下面的struct free_area,我们知道struct free_area是上图整体的一个结构,然后我们按照order可以找到指定的free_list结构,这个结构链接了所有系统中空闲page。

找到这个page,则将这个page从list上面删除,然后rmv_page_order()可也将page的标志位PG_buddy位删除,表示这个页不包含于buddy system。

将这个page所在的nr_free自减,之后调用expand()函数,这个函数存在的意义就是在高阶order分配了一个较小的page,但同时低阶order又没有合适的页分配。

static inline void expand(struct zone *zone, struct page *page,
    int low, int high, struct free_area *area,
    int migratetype)
{
    unsigned long size = 1 << high;

    while (high > low) {
        area--;
        high--;
        size >>= 1;
.....
        list_add(&page[size].lru, &area->free_list[migratetype]);
        area->nr_free++;
        set_page_order(&page[size], high);
    }
}

进入到expand()函数,这个函数是从current_order到请求order倒序遍历,函数的核心是通过current_order折半分裂,分裂完之后,将空闲的page挂到相同的order阶上面的free_area,然后nr_free++,set_page_order()作用是对于回收到伙伴系统的的内存的一个struct page实例中的private设置分配阶,并且设置PG_buddy。

 

完成伙伴系统的高阶page分裂。

 

PLKA: Page190

物理内存管理:伙伴系统数据结构分析