xxay/hifiasm
1
0
Fork 0
mirror of https://github.com/chhylp123/hifiasm.git synced 2026-09-29 06:58:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

r373

Browse Source
This commit is contained in:
chhylp123
2021-09-10 17:17:44 -04:00
parent 17b5cb21d3
commit 809ca18dda
18 changed files with 1569 additions and 1031 deletions
Download Patch File Download Diff File Expand all files Collapse all files
htab.cpp
+299 -331
View File
@@ -38,75 +38,6 @@ void *ha_flt_tab_hp;
ha_pt_t *ha_idx_hp;
void *ha_ct_table;
#define MZ_FUNC_INIT(sf, HType) \
static inline void sf##_init_kuf(pl_data_t *p, st_data_t *s){\
int i, n_pre = 1<<p->opt->pre, m;\
/**allocate the k-mer buffer**/\
CALLOC(s->buf, n_pre);\
m = (int)(s->nk * 1.2 / n_pre) + 1;\
/**pre-allocate memory for each of 4096 buffer**/\
for (i = 0; i < n_pre; ++i) {\
s->buf[i].m = m;\
/**for 0-th counting, p->pt = NULL**/\
if (p->pt && !(p->flag&HAF_COUNT_REFINE)) MALLOC(s->buf[i].b_##sf, m);\
else MALLOC(s->buf[i].a, m);\
}\
}\
static inline void sf##_destory_kuf(pl_data_t *p, st_data_t *s, int n){\
int i;\
uint64_t n_ins = 0;\
/**n_ins is number of distinct k-mers**/\
for (i = 0; i < n; ++i) {\
n_ins += s->buf[i].n_ins;\
if (p->pt && !(p->flag&HAF_COUNT_REFINE)) free(s->buf[i].b_##sf);\
else free(s->buf[i].a);\
}\
if (p->ct) p->ct->tot += n_ins, p->ct->bs += s->sum_len;\
if (p->pt) p->pt->tot_pos += n_ins;\
free(s->buf);\
/**#if 0\
fprintf(stderr, "[M::%s::%.3f*%.2f] processed %ld sequences; %ld %s in the hash table\n", __func__,\
yak_realtime(), yak_cpu_usage(), (long)s->n_seq0 + s->n_seq,\
(long)(p->pt? p->pt->tot_pos : p->ct->tot), p->pt? "positions" : "distinct k-mers");\
#endif**/\
free(s);\
}\
static inline void sf##_pt_insert_buf(ch_buf_t *buf, int p, const HType *y){\
/**assign minimizer to one of 4096 bins by low 12 bits**/\
int pre = y->x & ((1<<p) - 1);\
ch_buf_t *b = &buf[pre];\
if (b->n == b->m) {\
b->m = b->m < 8? 8 : b->m + (b->m>>1);\
REALLOC(b->b_##sf, b->m);\
}\
b->b_##sf[b->n++] = *y;\
}\
static inline void sf##_mselect(pl_data_t *p, st_data_t *s){\
int i; uint32_t j;\
/**s->n_seq is how many reads at this buffer**/\
/**s->mz && s->mz_buf are lists of minimzer vectors**/\
CALLOC(s->sf, s->n_seq), CALLOC(s->sf##_buf, p->opt->n_thread), CALLOC(s->mt, p->opt->n_thread);\
/**calculate minimzers for each read, each read corresponds to one thread**/\
kt_for(p->opt->n_thread, worker_for_mz, s, s->n_seq);\
for (i = 0; i < p->opt->n_thread; ++i) free(s->mt[i].a), free(s->sf##_buf[i].a);\
free(s->mt), free(s->sf##_buf);\
/**insert minimizers**/\
if (p->pt && !(p->flag&HAF_COUNT_REFINE)) {/**insert whole minimizer**/\
for (i = 0; i < s->n_seq; ++i)\
for (j = 0; j < s->sf[i].n; ++j)\
sf##_pt_insert_buf(s->buf, p->opt->pre, &s->sf[i].a[j]);\
} else {/**just insert the hash key of minimizer**/\
for (i = 0; i < s->n_seq; ++i)\
for (j = 0; j < s->sf[i].n; ++j)\
ct_insert_buf(s->buf, p->opt->pre, s->sf[i].a[j].x);\
}\
for (i = 0; i < s->n_seq; ++i) {\
p->n_mz += s->sf[i].n;\
free(s->sf[i].a);\
if (!p->is_store) free(s->seq[i]);\
}\
free(s->sf);}
/***************************
* Yak specific parameters *
***************************/
@@ -601,77 +532,7 @@ const int ha_pt_cnt(const ha_pt_t *h, uint64_t hash)
/**********************************
* Buffer for counting all k-mers *
**********************************/
typedef struct {
int n, m;
uint64_t n_ins;
uint64_t *a;
ha_mz1_t *b_mz;
ha_mzl_t *b_mzl;
} ch_buf_t;
///p = 12
static inline void ct_insert_buf(ch_buf_t *buf, int p, uint64_t y) // insert a k-mer $y to a linear buffer
{
///assign k-mer to one of the 4096 bins
///using low 12 bits for assigning
///so all elements at b have the same low 12 bits
int pre = y & ((1<<p) - 1);
ch_buf_t *b = &buf[pre];
if (b->n == b->m) {
b->m = b->m < 8? 8 : b->m + (b->m>>1);
REALLOC(b->a, b->m);
}
b->a[b->n++] = y;
}
///buf is the read block, k is the k-mer length, p = 12, len is the read length, seq is the read
static void count_seq_buf(ch_buf_t *buf, int k, int p, int len, const char *seq) // insert k-mers in $seq to linear buffer $buf
{
int i, l;
uint64_t x[4], mask = (1ULL<<k) - 1, shift = k - 1;
for (i = l = 0, x[0] = x[1] = x[2] = x[3] = 0; i < len; ++i) {
int c = seq_nt4_table[(uint8_t)seq[i]];
///c = 00, 01, 10, 11
if (c < 4) { // not an "N" base
///x[0] & x[1] are the forward k-mer
///x[2] & x[3] are the reverse complementary k-mer
x[0] = (x[0] << 1 | (c&1)) & mask;
x[1] = (x[1] << 1 | (c>>1)) & mask;
x[2] = x[2] >> 1 | (uint64_t)(1 - (c&1)) << shift;
x[3] = x[3] >> 1 | (uint64_t)(1 - (c>>1)) << shift;
if (++l >= k)
ct_insert_buf(buf, p, yak_hash_long(x));
} else l = 0, x[0] = x[1] = x[2] = x[3] = 0; // if there is an "N", restart
}
}
static void count_seq_buf_HPC(ch_buf_t *buf, int k, int p, int len, const char *seq) // insert k-mers in $seq to linear buffer $buf
{
int i, l, last = -1;
uint64_t x[4], mask = (1ULL<<k) - 1, shift = k - 1;
for (i = l = 0, x[0] = x[1] = x[2] = x[3] = 0; i < len; ++i) {
int c = seq_nt4_table[(uint8_t)seq[i]];
if (c < 4) { // not an "N" base
if (c != last) {
x[0] = (x[0] << 1 | (c&1)) & mask;
x[1] = (x[1] << 1 | (c>>1)) & mask;
x[2] = x[2] >> 1 | (uint64_t)(1 - (c&1)) << shift;
x[3] = x[3] >> 1 | (uint64_t)(1 - (c>>1)) << shift;
if (++l >= k)
ct_insert_buf(buf, p, yak_hash_long(x));
last = c;
}
} else l = 0, last = -1, x[0] = x[1] = x[2] = x[3] = 0; // if there is an "N", restart
}
}
/******************
* K-mer counting *
******************/
KSEQ_INIT(gzFile, gzread)
#define HAF_COUNT_EXACT 0x1
#define HAF_COUNT_ALL 0x2
#define HAF_RS_WRITE_LEN 0x4
@@ -696,178 +557,285 @@ typedef struct { // global data structure for kt_pipeline()
const ma_utg_v *us_in;
} pl_data_t;
typedef struct { // data structure for each step in kt_pipeline()
pl_data_t *p;
uint64_t n_seq0; ///the start index of current buffer block at R_INF
///sum_len = total bases, nk = number of k-mers
int n_seq, m_seq, sum_len, nk, uq;
int *len;
char **seq;
ha_mz1_v *mz_buf;
ha_mz1_v *mz;
ha_mzl_v *mzl_buf;
ha_mzl_v *mzl;
ch_buf_t *buf;
st_mt_t *mt;
} st_data_t;
static void worker_for_insert(void *data, long i, int tid) // callback for kt_for()
{
st_data_t *s = (st_data_t*)data;
ch_buf_t *b = &s->buf[i];
if (s->p->pt)
{
if(s->p->flag&HAF_COUNT_REFINE) b->n_ins += ha_pt_cnt_insert_list(s->p->pt, b->n, b->a);
else b->n_ins += ha_pt_insert_list(s->p->pt, b->n, b->b_mz);
}
else///for 0-th count, go into here
{
b->n_ins += ha_ct_insert_list(s->p->ct, s->p->create_new, b->n, b->a);
}
#define MZ_TEST_INIT(sf, HType, VType, IType, Ia) \
typedef struct {int n, m; uint64_t n_ins; uint64_t *a; HType *b;} sf##_ch_buf_t;\
static inline void sf##_ct_insert_buf(sf##_ch_buf_t *buf, int p, uint64_t y) /** insert a k-mer $y to a linear buffer**/\
{\
/**assign k-mer to one of the 4096 bins**/\
/**using low 12 bits for assigning**/\
/**so all elements at b have the same low 12 bits**/\
int pre = y & ((1<<p) - 1);\
sf##_ch_buf_t *b = &buf[pre];\
if (b->n == b->m) {\
b->m = b->m < 8? 8 : b->m + (b->m>>1);\
REALLOC(b->a, b->m);\
}\
b->a[b->n++] = y;\
}\
/**buf is the read block, k is the k-mer length, p = 12, len is the read length, seq is the read**/\
static void sf##_count_seq_buf(sf##_ch_buf_t *buf, int k, int p, int len, const char *seq) /**insert k-mers in $seq to linear buffer $buf**/\
{\
int i, l;\
uint64_t x[4], mask = (1ULL<<k) - 1, shift = k - 1;\
for (i = l = 0, x[0] = x[1] = x[2] = x[3] = 0; i < len; ++i) {\
int c = seq_nt4_table[(uint8_t)seq[i]];\
/**c = 00, 01, 10, 11**/\
if (c < 4) { /** not an "N" base**/\
/**x[0] & x[1] are the forward k-mer**/\
/**x[2] & x[3] are the reverse complementary k-mer**/\
x[0] = (x[0] << 1 | (c&1)) & mask;\
x[1] = (x[1] << 1 | (c>>1)) & mask;\
x[2] = x[2] >> 1 | (uint64_t)(1 - (c&1)) << shift;\
x[3] = x[3] >> 1 | (uint64_t)(1 - (c>>1)) << shift;\
if (++l >= k)\
sf##_ct_insert_buf(buf, p, yak_hash_long(x));\
} else l = 0, x[0] = x[1] = x[2] = x[3] = 0; /** if there is an "N", restart**/\
}\
}\
static void sf##_count_seq_buf_HPC(sf##_ch_buf_t *buf, int k, int p, int len, const char *seq) /**insert k-mers in $seq to linear buffer $buf**/\
{\
int i, l, last = -1;\
uint64_t x[4], mask = (1ULL<<k) - 1, shift = k - 1;\
for (i = l = 0, x[0] = x[1] = x[2] = x[3] = 0; i < len; ++i) {\
int c = seq_nt4_table[(uint8_t)seq[i]];\
if (c < 4) { /** not an "N" base**/\
if (c != last) {\
x[0] = (x[0] << 1 | (c&1)) & mask;\
x[1] = (x[1] << 1 | (c>>1)) & mask;\
x[2] = x[2] >> 1 | (uint64_t)(1 - (c&1)) << shift;\
x[3] = x[3] >> 1 | (uint64_t)(1 - (c>>1)) << shift;\
if (++l >= k)\
sf##_ct_insert_buf(buf, p, yak_hash_long(x));\
last = c;\
}\
} else l = 0, last = -1, x[0] = x[1] = x[2] = x[3] = 0; /**if there is an "N", restart**/\
}\
}\
int sf##_ha_pt_insert_list(ha_pt_t *h, int n, const HType *a)\
{\
int j, mask = (1<<h->pre) - 1, n_ins = 0;\
ha_pt1_t *g;\
if (n == 0) return 0;\
g = &h->h[a[0].x&mask];\
for (j = 0; j < n; ++j) {\
uint64_t x = a[j].x >> h->pre;\
khint_t k;\
int n;\
IType *p;\
assert((a[j].x&mask) == (a[0].x&mask));\
k = yak_pt_get(g->h, x<<YAK_COUNTER_BITS);\
if (k == kh_end(g->h)) continue; \
n = kh_key(g->h, k) & YAK_MAX_COUNT;\
assert(n < YAK_MAX_COUNT);\
p = &g->Ia[kh_val(g->h, k) + n];\
p->rid = a[j].rid, p->rev = a[j].rev, p->pos = a[j].pos, p->span = a[j].span;\
/**(uint64_t)a[j].rid<<36 | (uint64_t)a[j].rev<<35 | (uint64_t)a[j].pos<<8 | (uint64_t)a[j].span;**/\
++kh_key(g->h, k);\
++n_ins;\
}\
return n_ins;\
}\
/** data structure for each step in kt_pipeline()**/\
typedef struct {pl_data_t *p;uint64_t n_seq0; int n_seq, m_seq, sum_len, nk, uq, *len; char **seq; VType *mz_buf; VType *mz;sf##_ch_buf_t *buf;st_mt_t *mt;} sf##_st_data_t;\
static void sf##_worker_for_insert(void *data, long i, int tid) /** callback for kt_for()**/\
{\
sf##_st_data_t *s = (sf##_st_data_t*)data;\
sf##_ch_buf_t *b = &s->buf[i];\
if (s->p->pt){\
if(s->p->flag&HAF_COUNT_REFINE) b->n_ins += ha_pt_cnt_insert_list(s->p->pt, b->n, b->a);\
else b->n_ins += sf##_ha_pt_insert_list(s->p->pt, b->n, b->b);\
}else{\
b->n_ins += ha_ct_insert_list(s->p->ct, s->p->create_new, b->n, b->a);\
}\
}\
static void sf##_worker_for_mz(void *data, long i, int tid)\
{\
sf##_st_data_t *s = (sf##_st_data_t*)data;\
/**get the corresponding minimzer vector of this read**/\
VType *b = &s->mz_buf[tid];\
s->mz_buf[tid].n = 0;\
sf##_ha_sketch(s->seq[i], s->len[i], s->p->opt->w, s->p->opt->k, s->n_seq0 + i, s->p->opt->is_HPC, b, s->p->flt_tab, asm_opt.mz_sample_dist, 0, 0, \
(s->p->pt&&(s->p->flag&HAF_COUNT_REFINE))?s->p->pt:NULL, s->p->opt->min_rcnt, asm_opt.dp_min_len, asm_opt.dp_e, &(s->mt[tid]), asm_opt.mz_rewin, s->uq);\
s->mz[i].n = s->mz[i].m = b->n;\
MALLOC(s->mz[i].a, b->n);\
memcpy(s->mz[i].a, b->a, b->n * sizeof(VType));\
}\
static inline void sf##_pt_insert_buf(sf##_ch_buf_t *buf, int p, const HType *y){\
/**assign minimizer to one of 4096 bins by low 12 bits**/\
int pre = y->x & ((1<<p) - 1);\
sf##_ch_buf_t *b = &buf[pre];\
if (b->n == b->m) {\
b->m = b->m < 8? 8 : b->m + (b->m>>1);\
REALLOC(b->b, b->m);\
}\
b->b[b->n++] = *y;\
}\
static void *sf##_worker_count(void *data, int step, void *in) /** callback for kt_pipeline()**/\
{\
pl_data_t *p = (pl_data_t*)data;\
if (step == 0) { /** step 1: read a block of sequences**/\
int ret;\
sf##_st_data_t *s;\
CALLOC(s, 1);\
s->p = p;\
s->n_seq0 = p->n_seq;\
if (p->rs_in && (p->flag & HAF_RS_READ)) {\
while (p->n_seq < p->rs_in->total_reads) {\
if ((p->flag & HAF_SKIP_READ) && p->rs_in->trio_flag[p->n_seq] != AMBIGU) {\
++p->n_seq;\
continue;\
}\
int l;\
recover_UC_Read(&p->ucr, p->rs_in, p->n_seq);\
l = p->ucr.length;\
if (s->n_seq == s->m_seq) {\
s->m_seq = s->m_seq < 16? 16 : s->m_seq + (s->m_seq>>1);\
REALLOC(s->len, s->m_seq);\
REALLOC(s->seq, s->m_seq);\
}\
MALLOC(s->seq[s->n_seq], l);\
memcpy(s->seq[s->n_seq], p->ucr.seq, l);\
s->len[s->n_seq++] = l;\
++p->n_seq;\
s->sum_len += l;\
s->nk += l >= p->opt->k? l - p->opt->k + 1 : 0;\
if (s->sum_len >= p->opt->chunk_size)\
break;\
}\
} else if(p->us_in) {\
ma_utg_t *u; s->uq = p->us_in->h;\
while (p->n_seq < p->us_in->n) {\
u = &(p->us_in->a[p->n_seq]);\
if (s->n_seq == s->m_seq) {\
s->m_seq = s->m_seq < 16? 16 : s->m_seq + (s->m_seq>>1);\
REALLOC(s->len, s->m_seq);\
REALLOC(s->seq, s->m_seq);\
}\
MALLOC(s->seq[s->n_seq], u->len);\
memcpy(s->seq[s->n_seq], u->s, u->len);\
s->len[s->n_seq++] = u->len;\
++p->n_seq;\
s->sum_len += u->len;\
s->nk += u->len >= p->opt->k? u->len - p->opt->k + 1 : 0;\
if (s->sum_len >= p->opt->chunk_size)\
break;\
}\
} else {\
while ((ret = kseq_read(p->ks)) >= 0) {\
int l = (int)(p->ks->seq.l) - (int)(p->opt->adaLen) - (int)(p->opt->adaLen);\
if(l <= 0) continue;\
if (p->n_seq >= 1<<28) {\
fprintf(stderr, "ERROR: this implementation supports no more than %d reads\n", 1<<28);\
exit(1);\
}\
if (p->rs_out) {\
/**for 0-th count, just insert read length to R_INF, instead of read**/\
if (p->flag & HAF_RS_WRITE_LEN) {\
assert(p->n_seq == p->rs_out->total_reads);\
ha_insert_read_len(p->rs_out, l, p->ks->name.l);\
} else if (p->flag & HAF_RS_WRITE_SEQ) {\
int i, n_N;\
assert(l == (int)p->rs_out->read_length[p->n_seq]);\
for (i = n_N = 0; i < l; ++i) /** count number of ambiguous bases**/\
if (seq_nt4_table[(uint8_t)p->ks->seq.s[i+p->opt->adaLen]] >= 4)\
++n_N;\
ha_compress_base(Get_READ(*p->rs_out, p->n_seq), p->ks->seq.s+p->opt->adaLen, l, &p->rs_out->N_site[p->n_seq], n_N);\
memcpy(&p->rs_out->name[p->rs_out->name_index[p->n_seq]], p->ks->name.s, p->ks->name.l);\
}\
}\
if (s->n_seq == s->m_seq) {\
s->m_seq = s->m_seq < 16? 16 : s->m_seq + (s->m_seq>>1);\
REALLOC(s->len, s->m_seq);\
REALLOC(s->seq, s->m_seq);\
}\
MALLOC(s->seq[s->n_seq], l);\
memcpy(s->seq[s->n_seq], p->ks->seq.s+p->opt->adaLen, l);\
s->len[s->n_seq++] = l;\
++p->n_seq;\
s->sum_len += l;\
s->nk += l >= p->opt->k? l - p->opt->k + 1 : 0;\
/**p->opt->chunk_size is the block max size**/\
if (s->sum_len >= p->opt->chunk_size)\
break;\
}\
}\
if (s->sum_len == 0) free(s);\
else return s;\
} else if (step == 1) { /** step 2: extract k-mers**/\
/**s is the block of reads**/\
sf##_st_data_t *s = (sf##_st_data_t*)in;\
int i, n_pre = 1<<p->opt->pre, m;\
/**allocate the k-mer buffer**/\
CALLOC(s->buf, n_pre);\
m = (int)(s->nk * 1.2 / n_pre) + 1;\
/**pre-allocate memory for each of 4096 buffer**/\
for (i = 0; i < n_pre; ++i) {\
s->buf[i].m = m;\
/**for 0-th counting, p->pt = NULL**/\
if (p->pt && !(p->flag&HAF_COUNT_REFINE)) MALLOC(s->buf[i].b, m);\
else MALLOC(s->buf[i].a, m);\
}\
if (p->opt->w == 1) { /** enumerate all k-mers**/\
int i;\
for (i = 0; i < s->n_seq; ++i) {\
if (p->opt->is_HPC)\
sf##_count_seq_buf_HPC(s->buf, p->opt->k, p->opt->pre, s->len[i], s->seq[i]);\
else\
sf##_count_seq_buf(s->buf, p->opt->k, p->opt->pre, s->len[i], s->seq[i]);\
if (!p->is_store) free(s->seq[i]);\
}\
} else { /** minimizers only**/\
uint32_t j;\
/**s->n_seq is how many reads at this buffer**/\
/**s->mz && s->mz_buf are lists of minimzer vectors**/\
CALLOC(s->mz, s->n_seq), CALLOC(s->mz_buf, p->opt->n_thread), CALLOC(s->mt, p->opt->n_thread);\
/**calculate minimzers for each read, each read corresponds to one thread**/\
kt_for(p->opt->n_thread, sf##_worker_for_mz, s, s->n_seq);\
for (i = 0; i < p->opt->n_thread; ++i) free(s->mt[i].a), free(s->mz_buf[i].a);\
free(s->mt), free(s->mz_buf);\
/**insert minimizers**/\
if (p->pt && !(p->flag&HAF_COUNT_REFINE)) {/**insert whole minimizer**/\
for (i = 0; i < s->n_seq; ++i)\
for (j = 0; j < s->mz[i].n; ++j)\
sf##_pt_insert_buf(s->buf, p->opt->pre, &s->mz[i].a[j]);\
} else {/**just insert the hash key of minimizer**/\
for (i = 0; i < s->n_seq; ++i)\
for (j = 0; j < s->mz[i].n; ++j)\
sf##_ct_insert_buf(s->buf, p->opt->pre, s->mz[i].a[j].x);\
}\
for (i = 0; i < s->n_seq; ++i) {\
p->n_mz += s->mz[i].n;\
free(s->mz[i].a);\
if (!p->is_store) free(s->seq[i]);\
}\
free(s->mz);\
}\
/**just clean seq**/\
free(s->seq); free(s->len);\
s->seq = 0, s->len = 0;\
return s;\
} else if (step == 2) { /** step 3: insert k-mers to hash table**/\
sf##_st_data_t *s = (sf##_st_data_t*)in;\
int i, n = 1<<p->opt->pre;uint64_t n_ins = 0;\
/**for 0-th counting, p->pt = NULL**/\
kt_for(p->opt->n_thread, sf##_worker_for_insert, s, n);\
/**n_ins is number of distinct k-mers**/\
for (i = 0; i < n; ++i) {\
n_ins += s->buf[i].n_ins;\
if (p->pt && !(p->flag&HAF_COUNT_REFINE)) free(s->buf[i].b);\
else free(s->buf[i].a);\
}\
if (p->ct) p->ct->tot += n_ins, p->ct->bs += s->sum_len;\
if (p->pt) p->pt->tot_pos += n_ins;\
free(s->buf);\
free(s);\
}\
return 0;\
}
static void worker_for_mz(void *data, long i, int tid)
{
st_data_t *s = (st_data_t*)data;
///get the corresponding minimzer vector of this read
ha_mz1_v *b = &s->mz_buf[tid];
s->mz_buf[tid].n = 0;
ha_sketch(s->seq[i], s->len[i], s->p->opt->w, s->p->opt->k, s->n_seq0 + i, s->p->opt->is_HPC, b, s->p->flt_tab, asm_opt.mz_sample_dist, 0, 0,
(s->p->pt&&(s->p->flag&HAF_COUNT_REFINE))?s->p->pt:NULL, s->p->opt->min_rcnt, asm_opt.dp_min_len, asm_opt.dp_e, &(s->mt[tid]), asm_opt.mz_rewin, s->uq);
s->mz[i].n = s->mz[i].m = b->n;
MALLOC(s->mz[i].a, b->n);
memcpy(s->mz[i].a, b->a, b->n * sizeof(ha_mz1_t));
}
MZ_TEST_INIT(mz1, ha_mz1_t, ha_mz1_v, ha_idxpos_t, a)
MZ_TEST_INIT(mz2, ha_mzl_t, ha_mzl_v, ha_idxposl_t, al)
MZ_FUNC_INIT(mz, ha_mz1_t)
MZ_FUNC_INIT(mzl, ha_mzl_t)
static void *worker_count(void *data, int step, void *in) // callback for kt_pipeline()
{
pl_data_t *p = (pl_data_t*)data;
if (step == 0) { // step 1: read a block of sequences
int ret;
st_data_t *s;
CALLOC(s, 1);
s->p = p;
s->n_seq0 = p->n_seq;
if (p->rs_in && (p->flag & HAF_RS_READ)) {
while (p->n_seq < p->rs_in->total_reads) {
if ((p->flag & HAF_SKIP_READ) && p->rs_in->trio_flag[p->n_seq] != AMBIGU) {
++p->n_seq;
continue;
}
int l;
recover_UC_Read(&p->ucr, p->rs_in, p->n_seq);
l = p->ucr.length;
if (s->n_seq == s->m_seq) {
s->m_seq = s->m_seq < 16? 16 : s->m_seq + (s->m_seq>>1);
REALLOC(s->len, s->m_seq);
REALLOC(s->seq, s->m_seq);
}
MALLOC(s->seq[s->n_seq], l);
memcpy(s->seq[s->n_seq], p->ucr.seq, l);
s->len[s->n_seq++] = l;
++p->n_seq;
s->sum_len += l;
s->nk += l >= p->opt->k? l - p->opt->k + 1 : 0;
if (s->sum_len >= p->opt->chunk_size)
break;
}
} else if(p->us_in) {
ma_utg_t *u; s->uq = 1;
while (p->n_seq < p->us_in->n) {
u = &(p->us_in->a[p->n_seq]);
if (s->n_seq == s->m_seq) {
s->m_seq = s->m_seq < 16? 16 : s->m_seq + (s->m_seq>>1);
REALLOC(s->len, s->m_seq);
REALLOC(s->seq, s->m_seq);
}
MALLOC(s->seq[s->n_seq], u->len);
memcpy(s->seq[s->n_seq], u->s, u->len);
s->len[s->n_seq++] = u->len;
++p->n_seq;
s->sum_len += u->len;
s->nk += u->len >= p->opt->k? u->len - p->opt->k + 1 : 0;
if (s->sum_len >= p->opt->chunk_size)
break;
}
} else {
while ((ret = kseq_read(p->ks)) >= 0) {
int l = (int)(p->ks->seq.l) - (int)(p->opt->adaLen) - (int)(p->opt->adaLen);
if(l <= 0) continue;
if (p->n_seq >= 1<<28) {
fprintf(stderr, "ERROR: this implementation supports no more than %d reads\n", 1<<28);
exit(1);
}
if (p->rs_out) {
///for 0-th count, just insert read length to R_INF, instead of read
if (p->flag & HAF_RS_WRITE_LEN) {
assert(p->n_seq == p->rs_out->total_reads);
ha_insert_read_len(p->rs_out, l, p->ks->name.l);
} else if (p->flag & HAF_RS_WRITE_SEQ) {
int i, n_N;
assert(l == (int)p->rs_out->read_length[p->n_seq]);
for (i = n_N = 0; i < l; ++i) // count number of ambiguous bases
if (seq_nt4_table[(uint8_t)p->ks->seq.s[i+p->opt->adaLen]] >= 4)
++n_N;
ha_compress_base(Get_READ(*p->rs_out, p->n_seq), p->ks->seq.s+p->opt->adaLen, l, &p->rs_out->N_site[p->n_seq], n_N);
memcpy(&p->rs_out->name[p->rs_out->name_index[p->n_seq]], p->ks->name.s, p->ks->name.l);
}
}
///for 0-th count, insert both seq and length to local block
if (s->n_seq == s->m_seq) {
s->m_seq = s->m_seq < 16? 16 : s->m_seq + (s->m_seq>>1);
REALLOC(s->len, s->m_seq);
REALLOC(s->seq, s->m_seq);
}
MALLOC(s->seq[s->n_seq], l);
memcpy(s->seq[s->n_seq], p->ks->seq.s+p->opt->adaLen, l);
s->len[s->n_seq++] = l;
++p->n_seq;
s->sum_len += l;
s->nk += l >= p->opt->k? l - p->opt->k + 1 : 0;
///p->opt->chunk_size is the block max size
if (s->sum_len >= p->opt->chunk_size)
break;
}
}
if (s->sum_len == 0) free(s);
else return s;
} else if (step == 1) { // step 2: extract k-mers
///s is the block of reads
st_data_t *s = (st_data_t*)in;
if(p->us_in) mzl_init_kuf(p, s);
else mz_init_kuf(p, s);
// fill the buffer
///for 0-th counting, p->opt->w == 1
if (p->opt->w == 1) { // enumerate all k-mers
///scan all reads
int i;
for (i = 0; i < s->n_seq; ++i) {
if (p->opt->is_HPC)
count_seq_buf_HPC(s->buf, p->opt->k, p->opt->pre, s->len[i], s->seq[i]);
else
count_seq_buf(s->buf, p->opt->k, p->opt->pre, s->len[i], s->seq[i]);
if (!p->is_store) free(s->seq[i]);
}
} else { // minimizers only
if(p->us_in) mzl_mselect(p, s);
else mz_mselect(p, s);
}
///just clean seq
free(s->seq); free(s->len);
s->seq = 0, s->len = 0;
return s;
} else if (step == 2) { // step 3: insert k-mers to hash table
st_data_t *s = (st_data_t*)in;
///for 0-th counting, p->pt = NULL
kt_for(p->opt->n_thread, worker_for_insert, s, 1<<p->opt->pre);
if(p->us_in) mzl_destory_kuf(p, s, 1<<p->opt->pre);
else mz_destory_kuf(p, s, 1<<p->opt->pre);
}
return 0;
}
void debug_adapter(const hifiasm_opt_t *asm_opt, All_reads *rs)
{
@@ -951,7 +919,8 @@ static ha_ct_t *yak_count(const yak_copt_t *opt, const char *fn, int flag, ha_pt
pl.ct = ha_ct_init(opt->k, opt->pre, opt->bf_n_hash, opt->bf_shift);
}
if(pl.ct) pl.ct->bs = 0;
kt_pipeline(3, worker_count, &pl, 3);
if(ug_rs) kt_pipeline(3, mz2_worker_count, &pl, 3);
else kt_pipeline(3, mz1_worker_count, &pl, 3);
if (read_rs) {
destory_UC_Read(&pl.ucr);
} else if(!read_rs && !ug_rs) {
@@ -963,7 +932,7 @@ static ha_ct_t *yak_count(const yak_copt_t *opt, const char *fn, int flag, ha_pt
return pl.ct;
}
ha_ct_t *ha_count(const hifiasm_opt_t *asm_opt, int flag, ha_pt_t *p0, const void *flt_tab, All_reads *rs, ma_utg_v *us, int keep_adapter, int *low_freq)
ha_ct_t *ha_count(const hifiasm_opt_t *asm_o, int flag, int HPC, int k, int w, ha_pt_t *p0, const void *flt_tab, All_reads *rs, ma_utg_v *us, int keep_adapter, int *low_freq)
{
int i;
int64_t n_seq = 0;
@@ -979,20 +948,19 @@ ha_ct_t *ha_count(const hifiasm_opt_t *asm_opt, int flag, ha_pt_t *p0, const voi
malloc_All_reads(rs);
}
yak_copt_init(&opt);
opt.k = us? asm_opt->ul_mer_length:asm_opt->k_mer_length;
///always 0
opt.is_HPC = !(asm_opt->flag&HA_F_NO_HPC);
opt.k = k;
opt.is_HPC = HPC;
///for ft-counting, shoud be 1
opt.w = flag & HAF_COUNT_ALL? 1 : (us? asm_opt->ul_mz_win:asm_opt->mz_win);
opt.w = flag & HAF_COUNT_ALL? 1 : w;
///for ft-counting, shoud be 37
///for ha_pt_gen, shoud be 0
opt.bf_shift = flag & HAF_COUNT_EXACT? 0 : asm_opt->bf_shift;
opt.n_thread = asm_opt->thread_num;
opt.adaLen = (keep_adapter? asm_opt->adapterLen : 0);
opt.bf_shift = flag & HAF_COUNT_EXACT? 0 : asm_o->bf_shift;
opt.n_thread = asm_o->thread_num;
opt.adaLen = (keep_adapter? asm_o->adapterLen : 0);
opt.min_rcnt = (low_freq?*low_freq:-1);
///asm_opt->num_reads is the number of fastq files
for (i = n_bs = 0; i < (us?1:asm_opt->num_reads); ++i){
h = yak_count(&opt, asm_opt->read_file_names[i], flag|HAF_CREATE_NEW, p0, h, flt_tab, rs, us, &n_seq);
for (i = n_bs = 0; i < (us?1:asm_o->num_reads); ++i){
h = yak_count(&opt, asm_o->read_file_names[i], flag|HAF_CREATE_NEW, p0, h, flt_tab, rs, us, &n_seq);
if(h) n_bs += h->bs;
}
if(h) h->bs = n_bs;
@@ -1077,32 +1045,32 @@ void debug_ct_index(void* q_ct_idx, void* r_ct_idx)
* High-level interfaces *
*************************/
void *ha_ft_ug_gen(const hifiasm_opt_t *asm_opt, ma_utg_v *us, int hap_n)
void *ha_ft_ug_gen(const hifiasm_opt_t *asm_opt, ma_utg_v *us, int is_HPC, int k, int w, int min_freq, int max_freq)
{
yak_ft_t *flt_tab;
ha_ct_t *h;
///HAF_COUNT_EXACT ---> no bf; HAF_COUNT_ALL ---> no minimizer
h = ha_count(asm_opt, HAF_COUNT_ALL|HAF_UG_READ|HAF_COUNT_EXACT, NULL, NULL, NULL, us, 0, NULL);
ha_ct_shrink(h, 1, YAK_MAX_COUNT-1, asm_opt->thread_num);
h = ha_count(asm_opt, HAF_COUNT_ALL|HAF_UG_READ|HAF_COUNT_EXACT, is_HPC, k, w, NULL, NULL, NULL, us, 0, NULL);
ha_ct_shrink(h, min_freq, max_freq>YAK_MAX_COUNT-1?YAK_MAX_COUNT-1:max_freq, asm_opt->thread_num);
flt_tab = gen_hh(h, asm_opt->max_kmer_cnt);
ha_ct_destroy(h);
return (void*)flt_tab;
}
ha_pt_t *ha_pt_ug_gen(const hifiasm_opt_t *asm_opt, const void *flt_tab, ma_utg_v *us, int hap_n)
ha_pt_t *ha_pt_ug_gen(const hifiasm_opt_t *asm_opt, const void *flt_tab, ma_utg_v *us, int is_HPC, int k, int w, int min_freq)
{
ha_ct_t *ct;
ha_pt_t *pt;
///HAF_COUNT_EXACT: no bf
ct = ha_count(asm_opt, HAF_COUNT_EXACT|HAF_UG_READ, NULL, flt_tab, NULL, us, 0, NULL);
ct = ha_count(asm_opt, HAF_COUNT_EXACT|HAF_UG_READ, is_HPC, k, w, NULL, flt_tab, NULL, us, 0, NULL);
fprintf(stderr, "[M::%s::%.3f*%.2f] ==> counted %ld distinct minimizer k-mers\n", __func__,
yak_realtime(), yak_cpu_usage(), (long)ct->tot);
///minimizer with YAK_MAX_COUNT occ may apper > YAK_MAX_COUNT times, so it may lead to overflow at ha_pt_gen
ha_ct_shrink(ct, 1, YAK_MAX_COUNT - 1, asm_opt->thread_num);
ha_ct_shrink(ct, min_freq, YAK_MAX_COUNT - 1, asm_opt->thread_num);
pt = ha_pt_gen(ct, asm_opt->thread_num, 1);
ha_count(asm_opt, HAF_COUNT_EXACT|HAF_UG_READ, pt, flt_tab, NULL, us, 0, NULL);
ha_count(asm_opt, HAF_COUNT_EXACT|HAF_UG_READ, is_HPC, k, w, pt, flt_tab, NULL, us, 0, NULL);
//ha_pt_sort(pt, asm_opt->thread_num);
fprintf(stderr, "[M::%s::%.3f*%.2f] ==> indexed %ld positions\n", __func__,
yak_realtime(), yak_cpu_usage(), (long)pt->tot_pos);
@@ -1116,7 +1084,7 @@ void *ha_ft_gen(const hifiasm_opt_t *asm_opt, All_reads *rs, int *hom_cov, int i
int peak_hom, peak_het, cutoff = YAK_MAX_COUNT - 1, ex_flag = 0;
if(is_hp_mode) ex_flag = HAF_RS_READ|HAF_SKIP_READ;
ha_ct_t *h;
h = ha_count(asm_opt, HAF_COUNT_ALL|HAF_RS_WRITE_LEN|ex_flag, NULL, NULL, rs, NULL, 1, NULL);
h = ha_count(asm_opt, HAF_COUNT_ALL|HAF_RS_WRITE_LEN|ex_flag, !(asm_opt->flag&HA_F_NO_HPC), asm_opt->k_mer_length, asm_opt->mz_win, NULL, NULL, rs, NULL, 1, NULL);
if((asm_opt->flag & HA_F_VERBOSE_GFA))
{
write_ct_index((void*)h, asm_opt->output_file_name);
@@ -1148,12 +1116,12 @@ ha_pt_t *ha_pt_gen_dp(const hifiasm_opt_t *asm_opt, ha_ct_t *ct, int flag, int n
{
int low_freq = mz_low_b(peak_hom, peak_het);
ha_pt_t *pt = ha_pt_gen_count(ct, n_thread); ///key = cnt, val = 0
ha_count(asm_opt, HAF_COUNT_EXACT|HAF_COUNT_REFINE|flag, pt, flt_tab, rs, NULL, 1, &low_freq);
ha_count(asm_opt, HAF_COUNT_EXACT|HAF_COUNT_REFINE|flag, !(asm_opt->flag&HA_F_NO_HPC), asm_opt->k_mer_length, asm_opt->mz_win, pt, flt_tab, rs, NULL, 1, &low_freq);
uint64_t occ = ha_pt_shrink(pt, n_thread);
if(flag&HAF_RS_WRITE_LEN) flag -= HAF_RS_WRITE_LEN;
if(flag&HAF_RS_WRITE_SEQ) flag -= HAF_RS_WRITE_SEQ;
flag |= HAF_RS_READ; pt->tot_pos = 0;
ha_count(asm_opt, HAF_COUNT_EXACT|flag, pt, flt_tab, rs, NULL, 1, NULL);
ha_count(asm_opt, HAF_COUNT_EXACT|flag, !(asm_opt->flag&HA_F_NO_HPC), asm_opt->k_mer_length, asm_opt->mz_win, pt, flt_tab, rs, NULL, 1, NULL);
// fprintf(stderr, "[M::%s::] counted %lu distinct minimizer k-mers\n", __func__, pt->tot);
// fprintf(stderr, "[M::%s::] collected %lu minimizers\n\n\n", __func__, pt->tot_pos);
assert(occ == pt->tot_pos);
@@ -1177,7 +1145,7 @@ ha_pt_t *ha_pt_gen(const hifiasm_opt_t *asm_opt, const void *flt_tab, int read_f
}
if(is_hp_mode) extra_flag1 |= HAF_SKIP_READ, extra_flag2 |= HAF_SKIP_READ;
ct = ha_count(asm_opt, HAF_COUNT_EXACT|extra_flag1, NULL, flt_tab, rs, NULL, 1, NULL);
ct = ha_count(asm_opt, HAF_COUNT_EXACT|extra_flag1, !(asm_opt->flag&HA_F_NO_HPC), asm_opt->k_mer_length, asm_opt->mz_win, NULL, flt_tab, rs, NULL, 1, NULL);
fprintf(stderr, "[M::%s::%.3f*%.2f] ==> counted %ld distinct minimizer k-mers\n", __func__,
yak_realtime(), yak_cpu_usage(), (long)ct->tot);
ha_ct_hist(ct, cnt, asm_opt->thread_num);
@@ -1203,7 +1171,7 @@ ha_pt_t *ha_pt_gen(const hifiasm_opt_t *asm_opt, const void *flt_tab, int read_f
{
fprintf(stderr, "[M::%s::] counting in normal mode\n", __func__);
pt = ha_pt_gen(ct, asm_opt->thread_num, 0);
ha_count(asm_opt, HAF_COUNT_EXACT|extra_flag2, pt, flt_tab, rs, NULL, 1, NULL);
ha_count(asm_opt, HAF_COUNT_EXACT|extra_flag2, !(asm_opt->flag&HA_F_NO_HPC), asm_opt->k_mer_length, asm_opt->mz_win, pt, flt_tab, rs, NULL, 1, NULL);
assert((uint64_t)tot_cnt == pt->tot_pos);
}
else