X-Git-Url: https://git.donarmstrong.com/?p=rsem.git;a=blobdiff_plain;f=Gibbs.cpp;h=b0287a334cef604ea4385e250c7e22a8511c0424;hp=979860bda97cf07bd6e2a4f37869fa59186c7e3a;hb=1c50ecf9ddc331100184b9c3f95b9d400e4e8303;hpb=946f9a6adb2a82048c8453d44693cd3838d32939 diff --git a/Gibbs.cpp b/Gibbs.cpp index 979860b..b0287a3 100644 --- a/Gibbs.cpp +++ b/Gibbs.cpp @@ -5,8 +5,10 @@ #include #include #include +#include #include "utils.h" +#include "my_assert.h" #include "sampling.h" #include "Model.h" @@ -20,6 +22,15 @@ using namespace std; +struct Params { + int no, nsamples; + FILE *fo; + engine_type *engine; + double *pme_c, *pve_c; //posterior mean and variance vectors on counts + double *pme_tpm, *pme_fpkm; +}; + + struct Item { int sid; double conprb; @@ -30,10 +41,14 @@ struct Item { } }; +int nThreads; + int model_type; -int m, M, N0, N1, nHits; +int m, M; +READ_INT_TYPE N0, N1; +HIT_INT_TYPE nHits; double totc; -int BURNIN, CHAINLEN, GAP; +int BURNIN, NSAMPLES, GAP; char imdName[STRLEN], statName[STRLEN]; char thetaF[STRLEN], ofgF[STRLEN], groupF[STRLEN], refF[STRLEN], modelF[STRLEN]; char cvsF[STRLEN]; @@ -41,15 +56,22 @@ char cvsF[STRLEN]; Refs refs; GroupInfo gi; -vector theta, pme_theta, pme_c, eel; - -vector s, z; +vector s; vector hits; -vector counts; +vector eel; +double *mw; + +vector pme_c, pve_c; //global posterior mean and variance vectors on counts +vector pme_tpm, pme_fpkm; + +bool var_opt; bool quiet; -vector arr; +Params *paramsArray; +pthread_t *threads; +pthread_attr_t attr; +int rc; void load_data(char* reference_name, char* statName, char* imdName) { ifstream fin; @@ -66,34 +88,12 @@ void load_data(char* reference_name, char* statName, char* imdName) { gi.load(groupF); m = gi.getm(); - //load thetaF - sprintf(thetaF, "%s.theta",statName); - fin.open(thetaF); - if (!fin.is_open()) { - fprintf(stderr, "Cannot open %s!\n", thetaF); - exit(-1); - } - fin>>tmpVal; - if (tmpVal != M + 1) { - fprintf(stderr, "Number of transcripts is not consistent in %s and %s!\n", refF, thetaF); - exit(-1); - } - theta.clear(); theta.resize(M + 1); - for (int i = 0; i <= M; i++) fin>>theta[i]; - fin.close(); - //load ofgF; sprintf(ofgF, "%s.ofg", imdName); fin.open(ofgF); - if (!fin.is_open()) { - fprintf(stderr, "Cannot open %s!\n", ofgF); - exit(-1); - } + general_assert(fin.is_open(), "Cannot open " + cstrtos(ofgF) + "!"); fin>>tmpVal>>N0; - if (tmpVal != M) { - fprintf(stderr, "M in %s is not consistent with %s!\n", ofgF, refF); - exit(-1); - } + general_assert(tmpVal == M, "M in " + cstrtos(ofgF) + " is not consistent with " + cstrtos(refF) + "!"); getline(fin, line); s.clear(); hits.clear(); @@ -113,237 +113,402 @@ void load_data(char* reference_name, char* statName, char* imdName) { N1 = s.size() - 1; nHits = hits.size(); + totc = N0 + N1 + (M + 1); + if (verbose) { printf("Loading Data is finished!\n"); } } -void init() { - int len, fr, to; +template +void calcExpectedEffectiveLengths(ModelType& model) { + int lb, ub, span; + double *pdf = NULL, *cdf = NULL, *clen = NULL; // clen[i] = sigma_{j=1}^{i}pdf[i]*(lb+i) + + model.getGLD().copyTo(pdf, cdf, lb, ub, span); + clen = new double[span + 1]; + clen[0] = 0.0; + for (int i = 1; i <= span; i++) { + clen[i] = clen[i - 1] + pdf[i] * (lb + i); + } - arr.clear(); - z.clear(); - counts.clear(); + eel.assign(M + 1, 0.0); + for (int i = 1; i <= M; i++) { + int totLen = refs.getRef(i).getTotLen(); + int fullLen = refs.getRef(i).getFullLen(); + int pos1 = max(min(totLen - fullLen + 1, ub) - lb, 0); + int pos2 = max(min(totLen, ub) - lb, 0); - z.resize(N1); - counts.resize(M + 1, 1); // 1 pseudo count - counts[0] += N0; + if (pos2 == 0) { eel[i] = 0.0; continue; } + + eel[i] = fullLen * cdf[pos1] + ((cdf[pos2] - cdf[pos1]) * (totLen + 1) - (clen[pos2] - clen[pos1])); + assert(eel[i] >= 0); + if (eel[i] < MINEEL) { eel[i] = 0.0; } + } + + delete[] pdf; + delete[] cdf; + delete[] clen; +} - for (int i = 0; i < N1; i++) { - fr = s[i]; to = s[i + 1]; - len = to - fr; - arr.resize(len); - for (int j = fr; j < to; j++) { - arr[j - fr] = theta[hits[j].sid] * hits[j].conprb; - if (j > fr) arr[j - fr] += arr[j - fr - 1]; // cumulative - } - z[i] = hits[fr + sample(arr, len)].sid; - ++counts[z[i]]; +template +void init_model_related(char* modelF) { + ModelType model; + model.read(modelF); + + calcExpectedEffectiveLengths(model); + memcpy(mw, model.getMW(), sizeof(double) * (M + 1)); // otherwise, after exiting this procedure, mw becomes undefined +} + +// assign threads +void init() { + int quotient, left; + char outF[STRLEN]; + + quotient = NSAMPLES / nThreads; + left = NSAMPLES % nThreads; + + sprintf(cvsF, "%s.countvectors", imdName); + paramsArray = new Params[nThreads]; + threads = new pthread_t[nThreads]; + + for (int i = 0; i < nThreads; i++) { + paramsArray[i].no = i; + + paramsArray[i].nsamples = quotient; + if (i < left) paramsArray[i].nsamples++; + + sprintf(outF, "%s%d", cvsF, i); + paramsArray[i].fo = fopen(outF, "w"); + + paramsArray[i].engine = engineFactory::new_engine(); + paramsArray[i].pme_c = new double[M + 1]; + memset(paramsArray[i].pme_c, 0, sizeof(double) * (M + 1)); + paramsArray[i].pve_c = new double[M + 1]; + memset(paramsArray[i].pve_c, 0, sizeof(double) * (M + 1)); + paramsArray[i].pme_tpm = new double[M + 1]; + memset(paramsArray[i].pme_tpm, 0, sizeof(double) * (M + 1)); + paramsArray[i].pme_fpkm = new double[M + 1]; + memset(paramsArray[i].pme_fpkm, 0, sizeof(double) * (M + 1)); } - totc = N0 + N1 + (M + 1); + /* set thread attribute to be joinable */ + pthread_attr_init(&attr); + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE); - if (verbose) { printf("Initialization is finished!\n"); } + if (verbose) { printf("Initialization finished!\n"); } } -void writeCountVector(FILE* fo) { +//sample theta from Dir(1) +void sampleTheta(engine_type& engine, vector& theta) { + gamma_dist gm(1); + gamma_generator gmg(engine, gm); + double denom; + + theta.assign(M + 1, 0); + denom = 0.0; + for (int i = 0; i <= M; i++) { + theta[i] = gmg(); + denom += theta[i]; + } + assert(denom > EPSILON); + for (int i = 0; i <= M; i++) theta[i] /= denom; +} + +void writeCountVector(FILE* fo, vector& counts) { for (int i = 0; i < M; i++) { fprintf(fo, "%d ", counts[i]); } fprintf(fo, "%d\n", counts[M]); } -void Gibbs(char* imdName) { - FILE *fo; - int fr, to, len; +void polishTheta(vector& theta, const vector& eel, const double* mw) { + double sum = 0.0; - sprintf(cvsF, "%s.countvectors", imdName); - fo = fopen(cvsF, "w"); - assert(CHAINLEN % GAP == 0); - fprintf(fo, "%d %d\n", CHAINLEN / GAP, M + 1); - //fprintf(fo, "%d %d\n", CHAINLEN, M + 1); + /* The reason that for noise gene, mw value is 1 is : + * currently, all masked positions are for poly(A) sites, which in theory should be filtered out. + * So the theta0 does not containing reads from any masked position + */ + + for (int i = 0; i <= M; i++) { + // i == 0, mw[i] == 1 + if (i > 0 && (mw[i] < EPSILON || eel[i] < EPSILON)) { + theta[i] = 0.0; + continue; + } + theta[i] = theta[i] / mw[i]; + sum += theta[i]; + } + // currently is OK, since no transcript should be masked totally, only the poly(A) tail related part will be masked + general_assert(sum >= EPSILON, "No effective length is no less than" + ftos(MINEEL, 6) + " !"); + for (int i = 0; i <= M; i++) theta[i] /= sum; +} + +void calcExpressionValues(const vector& theta, const vector& eel, vector& tpm, vector& fpkm) { + double denom; + vector frac; + + //calculate fraction of count over all mappabile reads + denom = 0.0; + frac.assign(M + 1, 0.0); + for (int i = 1; i <= M; i++) + if (eel[i] >= EPSILON) { + frac[i] = theta[i]; + denom += frac[i]; + } + general_assert(denom >= EPSILON, "No alignable reads?!"); + for (int i = 1; i <= M; i++) frac[i] /= denom; + + //calculate FPKM + fpkm.assign(M + 1, 0.0); + for (int i = 1; i <= M; i++) + if (eel[i] >= EPSILON) fpkm[i] = frac[i] * 1e9 / eel[i]; + + //calculate TPM + tpm.assign(M + 1, 0.0); + denom = 0.0; + for (int i = 1; i <= M; i++) denom += fpkm[i]; + for (int i = 1; i <= M; i++) tpm[i] = fpkm[i] / denom * 1e6; +} + +void* Gibbs(void* arg) { + int CHAINLEN; + HIT_INT_TYPE len, fr, to; + Params *params = (Params*)arg; + + vector theta, tpm, fpkm; + vector z, counts; + vector arr; + + uniform01 rg(*params->engine); + + // generate initial state + sampleTheta(*params->engine, theta); - pme_c.clear(); pme_c.resize(M + 1, 0.0); - pme_theta.clear(); pme_theta.resize(M + 1, 0.0); + z.assign(N1, 0); + + counts.assign(M + 1, 1); // 1 pseudo count + counts[0] += N0; + + for (READ_INT_TYPE i = 0; i < N1; i++) { + fr = s[i]; to = s[i + 1]; + len = to - fr; + arr.assign(len, 0); + for (HIT_INT_TYPE j = fr; j < to; j++) { + arr[j - fr] = theta[hits[j].sid] * hits[j].conprb; + if (j > fr) arr[j - fr] += arr[j - fr - 1]; // cumulative + } + z[i] = hits[fr + sample(rg, arr, len)].sid; + ++counts[z[i]]; + } + + // Gibbs sampling + CHAINLEN = 1 + (params->nsamples - 1) * GAP; for (int ROUND = 1; ROUND <= BURNIN + CHAINLEN; ROUND++) { - for (int i = 0; i < N1; i++) { + for (READ_INT_TYPE i = 0; i < N1; i++) { --counts[z[i]]; fr = s[i]; to = s[i + 1]; len = to - fr; - arr.resize(len); - for (int j = fr; j < to; j++) { + arr.assign(len, 0); + for (HIT_INT_TYPE j = fr; j < to; j++) { arr[j - fr] = counts[hits[j].sid] * hits[j].conprb; if (j > fr) arr[j - fr] += arr[j - fr - 1]; //cumulative } - z[i] = hits[fr + sample(arr, len)].sid; + z[i] = hits[fr + sample(rg, arr, len)].sid; ++counts[z[i]]; } if (ROUND > BURNIN) { - if ((ROUND - BURNIN - 1) % GAP == 0) writeCountVector(fo); - for (int i = 0; i <= M; i++) { - pme_c[i] += counts[i] - 1; - pme_theta[i] += counts[i] / totc; + if ((ROUND - BURNIN - 1) % GAP == 0) { + writeCountVector(params->fo, counts); + for (int i = 0; i <= M; i++) theta[i] = counts[i] / totc; + polishTheta(theta, eel, mw); + calcExpressionValues(theta, eel, tpm, fpkm); + for (int i = 0; i <= M; i++) { + params->pme_c[i] += counts[i] - 1; + params->pve_c[i] += (counts[i] - 1) * (counts[i] - 1); + params->pme_tpm[i] += tpm[i]; + params->pme_fpkm[i] += fpkm[i]; + } } } - if (verbose) { printf("ROUND %d is finished!\n", ROUND); } + if (verbose && ROUND % 100 == 0) { printf("Thread %d, ROUND %d is finished!\n", params->no, ROUND); } } - fclose(fo); - for (int i = 0; i <= M; i++) { - pme_c[i] /= CHAINLEN; - pme_theta[i] /= CHAINLEN; + return NULL; +} + +void release() { +// char inpF[STRLEN], command[STRLEN]; + string line; + + /* destroy attribute */ + pthread_attr_destroy(&attr); + delete[] threads; + + pme_c.assign(M + 1, 0); + pve_c.assign(M + 1, 0); + pme_tpm.assign(M + 1, 0); + pme_fpkm.assign(M + 1, 0); + for (int i = 0; i < nThreads; i++) { + fclose(paramsArray[i].fo); + delete paramsArray[i].engine; + for (int j = 0; j <= M; j++) { + pme_c[j] += paramsArray[i].pme_c[j]; + pve_c[j] += paramsArray[i].pve_c[j]; + pme_tpm[j] += paramsArray[i].pme_tpm[j]; + pme_fpkm[j] += paramsArray[i].pme_fpkm[j]; + } + delete[] paramsArray[i].pme_c; + delete[] paramsArray[i].pve_c; + delete[] paramsArray[i].pme_tpm; + delete[] paramsArray[i].pme_fpkm; } + delete[] paramsArray; - if (verbose) { printf("Gibbs is finished!\n"); } -} -template -void calcExpectedEffectiveLengths(ModelType& model) { - int lb, ub, span; - double *pdf = NULL, *cdf = NULL, *clen = NULL; // clen[i] = sigma_{j=1}^{i}pdf[i]*(lb+i) - - model.getGLD().copyTo(pdf, cdf, lb, ub, span); - clen = new double[span + 1]; - clen[0] = 0.0; - for (int i = 1; i <= span; i++) { - clen[i] = clen[i - 1] + pdf[i] * (lb + i); - } - - eel.clear(); - eel.resize(M + 1, 0.0); - for (int i = 1; i <= M; i++) { - int totLen = refs.getRef(i).getTotLen(); - int fullLen = refs.getRef(i).getFullLen(); - int pos1 = max(min(totLen - fullLen + 1, ub) - lb, 0); - int pos2 = max(min(totLen, ub) - lb, 0); - - if (pos2 == 0) { eel[i] = 0.0; continue; } - - eel[i] = fullLen * cdf[pos1] + ((cdf[pos2] - cdf[pos1]) * (totLen + 1) - (clen[pos2] - clen[pos1])); - assert(eel[i] >= 0); - if (eel[i] < MINEEL) { eel[i] = 0.0; } - } - - delete[] pdf; - delete[] cdf; - delete[] clen; + for (int i = 0; i <= M; i++) { + pme_c[i] /= NSAMPLES; + pve_c[i] = (pve_c[i] - NSAMPLES * pme_c[i] * pme_c[i]) / (NSAMPLES - 1); + pme_tpm[i] /= NSAMPLES; + pme_fpkm[i] /= NSAMPLES; + } } -template -void writeEstimatedParameters(char* modelF, char* imdName) { - ModelType model; - double denom; +void writeResults(char* imdName) { char outF[STRLEN]; FILE *fo; - model.read(modelF); - - calcExpectedEffectiveLengths(model); - - denom = pme_theta[0]; - for (int i = 1; i <= M; i++) - if (eel[i] < EPSILON) pme_theta[i] = 0.0; - else denom += pme_theta[i]; - if (denom <= 0) { fprintf(stderr, "No Expected Effective Length is no less than %.6g?!\n", MINEEL); exit(-1); } - for (int i = 0; i <= M; i++) pme_theta[i] /= denom; - - denom = 0.0; - double *mw = model.getMW(); - for (int i = 0; i <= M; i++) { - pme_theta[i] = (mw[i] < EPSILON ? 0.0 : pme_theta[i] / mw[i]); - denom += pme_theta[i]; - } - assert(denom >= EPSILON); - for (int i = 0; i <= M; i++) pme_theta[i] /= denom; + vector isopct; + vector gene_counts, gene_tpm, gene_fpkm; - //calculate tau values - double *tau = new double[M + 1]; - memset(tau, 0, sizeof(double) * (M + 1)); + //calculate IsoPct, etc. + isopct.assign(M + 1, 0.0); + gene_counts.assign(m, 0.0); gene_tpm.assign(m, 0.0); gene_fpkm.assign(m, 0.0); - denom = 0.0; - for (int i = 1; i <= M; i++) - if (eel[i] > EPSILON) { - tau[i] = pme_theta[i] / eel[i]; - denom += tau[i]; - } - if (denom <= 0) { fprintf(stderr, "No alignable reads?!\n"); exit(-1); } - //assert(denom > 0); - for (int i = 1; i <= M; i++) { - tau[i] /= denom; + for (int i = 0; i < m; i++) { + int b = gi.spAt(i), e = gi.spAt(i + 1); + for (int j = b; j < e; j++) { + gene_counts[i] += pme_c[j]; + gene_tpm[i] += pme_tpm[j]; + gene_fpkm[i] += pme_fpkm[j]; + } + if (gene_tpm[i] < EPSILON) continue; + for (int j = b; j < e; j++) + isopct[j] = pme_tpm[j] / gene_tpm[i]; } //isoform level results sprintf(outF, "%s.iso_res", imdName); fo = fopen(outF, "a"); - if (fo == NULL) { fprintf(stderr, "Cannot open %s!\n", outF); exit(-1); } + general_assert(fo != NULL, "Cannot open " + cstrtos(outF) + "!"); + for (int i = 1; i <= M; i++) fprintf(fo, "%.2f%c", pme_c[i], (i < M ? '\t' : '\n')); for (int i = 1; i <= M; i++) - fprintf(fo, "%.15g%c", tau[i], (i < M ? '\t' : '\n')); + fprintf(fo, "%.2f%c", pme_tpm[i], (i < M ? '\t' : '\n')); + for (int i = 1; i <= M; i++) + fprintf(fo, "%.2f%c", pme_fpkm[i], (i < M ? '\t' : '\n')); + for (int i = 1; i <= M; i++) + fprintf(fo, "%.2f%c", isopct[i] * 1e2, (i < M ? '\t' : '\n')); fclose(fo); //gene level results sprintf(outF, "%s.gene_res", imdName); fo = fopen(outF, "a"); - if (fo == NULL) { fprintf(stderr, "Cannot open %s!\n", outF); exit(-1); } - for (int i = 0; i < m; i++) { - double sumC = 0.0; // sum of pme counts - int b = gi.spAt(i), e = gi.spAt(i + 1); - for (int j = b; j < e; j++) { - sumC += pme_c[j]; - } - fprintf(fo, "%.15g%c", sumC, (i < m - 1 ? '\t' : '\n')); - } - for (int i = 0; i < m; i++) { - double sumT = 0.0; // sum of tau values - int b = gi.spAt(i), e = gi.spAt(i + 1); - for (int j = b; j < e; j++) { - sumT += tau[j]; - } - fprintf(fo, "%.15g%c", sumT, (i < m - 1 ? '\t' : '\n')); - } + general_assert(fo != NULL, "Cannot open " + cstrtos(outF) + "!"); + + for (int i = 0; i < m; i++) + fprintf(fo, "%.2f%c", gene_counts[i], (i < m - 1 ? '\t' : '\n')); + for (int i = 0; i < m; i++) + fprintf(fo, "%.2f%c", gene_tpm[i], (i < m - 1 ? '\t' : '\n')); + for (int i = 0; i < m; i++) + fprintf(fo, "%.2f%c", gene_fpkm[i], (i < m - 1 ? '\t' : '\n')); fclose(fo); - delete[] tau; - if (verbose) { printf("Gibbs based expression values are written!\n"); } } - int main(int argc, char* argv[]) { if (argc < 7) { - printf("Usage: rsem-run-gibbs reference_name sample_name sampleToken BURNIN CHAINLEN GAP [-q]\n"); + printf("Usage: rsem-run-gibbs reference_name imdName statName BURNIN NSAMPLES GAP [-p #Threads] [--var] [-q]\n"); exit(-1); } + strcpy(imdName, argv[2]); + strcpy(statName, argv[3]); + BURNIN = atoi(argv[4]); - CHAINLEN = atoi(argv[5]); + NSAMPLES = atoi(argv[5]); GAP = atoi(argv[6]); - sprintf(imdName, "%s.temp/%s", argv[2], argv[3]); - sprintf(statName, "%s.stat/%s", argv[2], argv[3]); - load_data(argv[1], statName, imdName); + nThreads = 1; + var_opt = false; quiet = false; - if (argc > 7 && !strcmp(argv[7], "-q")) { - quiet = true; + + for (int i = 7; i < argc; i++) { + if (!strcmp(argv[i], "-p")) nThreads = atoi(argv[i + 1]); + if (!strcmp(argv[i], "--var")) var_opt = true; + if (!strcmp(argv[i], "-q")) quiet = true; } verbose = !quiet; - init(); - Gibbs(imdName); + assert(NSAMPLES > 1); // Otherwise, we cannot calculate posterior variance + + if (nThreads > NSAMPLES) { + nThreads = NSAMPLES; + printf("Warning: Number of samples is less than number of threads! Change the number of threads to %d!\n", nThreads); + } + + load_data(argv[1], statName, imdName); sprintf(modelF, "%s.model", statName); FILE *fi = fopen(modelF, "r"); - if (fi == NULL) { fprintf(stderr, "Cannot open %s!\n", modelF); exit(-1); } - fscanf(fi, "%d", &model_type); + general_assert(fi != NULL, "Cannot open " + cstrtos(modelF) + "!"); + assert(fscanf(fi, "%d", &model_type) == 1); fclose(fi); + mw = new double[M + 1]; // make an extra copy + switch(model_type) { - case 0 : writeEstimatedParameters(modelF, imdName); break; - case 1 : writeEstimatedParameters(modelF, imdName); break; - case 2 : writeEstimatedParameters(modelF, imdName); break; - case 3 : writeEstimatedParameters(modelF, imdName); break; + case 0 : init_model_related(modelF); break; + case 1 : init_model_related(modelF); break; + case 2 : init_model_related(modelF); break; + case 3 : init_model_related(modelF); break; } + if (verbose) printf("Gibbs started!\n"); + + init(); + for (int i = 0; i < nThreads; i++) { + rc = pthread_create(&threads[i], &attr, Gibbs, (void*)(¶msArray[i])); + pthread_assert(rc, "pthread_create", "Cannot create thread " + itos(i) + " (numbered from 0)!"); + } + for (int i = 0; i < nThreads; i++) { + rc = pthread_join(threads[i], NULL); + pthread_assert(rc, "pthread_join", "Cannot join thread " + itos(i) + " (numbered from 0)!"); + } + release(); + + if (verbose) printf("Gibbs finished!\n"); + + writeResults(imdName); + + if (var_opt) { + char varF[STRLEN]; + + sprintf(varF, "%s.var", statName); + FILE *fo = fopen(varF, "w"); + general_assert(fo != NULL, "Cannot open " + cstrtos(varF) + "!"); + for (int i = 0; i < m; i++) { + int b = gi.spAt(i), e = gi.spAt(i + 1), number_of_isoforms = e - b; + for (int j = b; j < e; j++) { + fprintf(fo, "%s\t%d\t%.15g\t%.15g\n", refs.getRef(j).getName().c_str(), number_of_isoforms, pme_c[j], pve_c[j]); + } + } + fclose(fo); + } + + delete mw; // delete the copy + return 0; }